fix: catch null byte error from pgstac

CHANGES.md

@@ -9,6 +9,7 @@
 * remove: `asset_bidx` and `asset_expression` options in favor of the new asset notation
 * remove: `tile-scale` path parameter in favor of `tilesize` query parameter in tiles endpoints
 * remove: support for `vrt://{asset_name}` assets
+* fix: catch `null byte` error from the database
 
 ## 2.1.0 (2026-03-05)
 

tests/test_collections.py

@@ -834,3 +834,16 @@ def test_bbox_validation(app) -> None:
         f"/collections/{collection_id}/tiles", params={"bbox": "180,-90,-180,90"}
     )
     assert invalid_values_response.status_code == 422
+
+
+def test_nullbyte_error(app) -> None:
+    """Raise 500 error when request contains null byte."""
+    response = app.get(
+        f"/collections/{collection_id}%00/info",
+    )
+    assert response.status_code == 500
+
+    response = app.get(
+        f"/collections/{collection_id}/tiles/WebMercatorQuad/15/8589/12849/assets?sortby=-gsd%00",
+    )
+    assert response.status_code == 500

titiler/pgstac/dependencies.py

@@ -1,6 +1,7 @@
 """titiler-pgstac dependencies."""
 
 import json
+import logging
 import re
 import warnings
 from dataclasses import dataclass, field
@@ -14,6 +15,7 @@
 from cachetools.keys import hashkey
 from cql2 import Expr
 from fastapi import HTTPException, Path, Query
+from psycopg import Cursor
 from psycopg import errors as pgErrors
 from psycopg.rows import class_row, dict_row
 from psycopg_pool import ConnectionPool
@@ -25,11 +27,13 @@
 from titiler.core.dependencies import DefaultDependency
 from titiler.core.validation import validate_json
 from titiler.pgstac import model
-from titiler.pgstac.errors import MosaicNotFoundError, ReadOnlyPgSTACError
+from titiler.pgstac.errors import BackendError, MosaicNotFoundError, ReadOnlyPgSTACError
 from titiler.pgstac.settings import CacheSettings, RetrySettings
 from titiler.pgstac.utils import retry
 from titiler.pgstac.validation import parse_and_validate_bbox, validate_filter
 
+logger = logging.getLogger(__name__)
+
 cache_config = CacheSettings()
 retry_config = RetrySettings()
 ttl_cache = TTLCache(maxsize=cache_config.maxsize, ttl=cache_config.ttl)  # type: ignore
@@ -45,6 +49,31 @@ def SearchIdParams(
     return search_id
 
 
+def register_search_in_db(
+    cursor: Cursor[Any],
+    search: model.PgSTACSearch,
+    metadata: model.Metadata,
+) -> model.Search:
+    """Register the search query in the database and return the Search info."""
+    cursor.row_factory = class_row(model.Search)  # type: ignore
+    try:
+        cursor.execute(
+            "SELECT * FROM search_query(%s, _metadata => %s);",
+            (
+                search.model_dump_json(by_alias=True, exclude_none=True),
+                metadata.model_dump_json(exclude_none=True),
+            ),
+        )
+        search_info = cast(model.Search, cursor.fetchone())
+    except pgErrors.DataError as e:
+        logger.exception(e)
+        raise BackendError(
+            f"Backend returned an error for mosaic with body {search.model_dump_json(by_alias=True, exclude_none=True)}"
+        ) from e
+
+    return search_info
+
+
 @cached(  # type: ignore
     ttl_cache,
     key=lambda pool, collection_id, ids, bbox, datetime, query, sortby, filter_expr, filter_lang: (
@@ -119,11 +148,18 @@ def get_collection_id(  # noqa: C901
     search = model.PgSTACSearch.model_validate(search_params)
     with pool.connection() as conn:
         with conn.cursor(row_factory=dict_row) as cursor:
-            cursor.execute(
-                "SELECT * FROM pgstac.get_collection(%s);",
-                (collection_id,),
-            )
-            collection = cursor.fetchone()["get_collection"]  # type: ignore [index]
+            try:
+                cursor.execute(
+                    "SELECT * FROM pgstac.get_collection(%s);",
+                    (collection_id,),
+                )
+                collection = cursor.fetchone()["get_collection"]  # type: ignore [index]
+            except pgErrors.DataError as e:
+                logger.exception(e)
+                raise BackendError(
+                    f"Backend returned an error for Collection `{collection_id}`"
+                ) from e
+
             if not collection:
                 raise MosaicNotFoundError(f"CollectionId `{collection_id}` not found")
 
@@ -182,15 +218,7 @@ def get_collection_id(  # noqa: C901
                 conn.rollback()
                 pass
 
-            cursor.row_factory = class_row(model.Search)  # type: ignore
-            cursor.execute(
-                "SELECT * FROM search_query(%s, _metadata => %s);",
-                (
-                    search.model_dump_json(by_alias=True, exclude_none=True),
-                    metadata.model_dump_json(exclude_none=True),
-                ),
-            )
-            search_info = cast(model.Search, cursor.fetchone())
+            search_info = register_search_in_db(cursor, search, metadata)
 
     return search_info.id
 

titiler/pgstac/errors.py

@@ -17,8 +17,13 @@ class NoLayerFound(TilerError):
     """Cannot find any valid Layer."""
 
 
+class BackendError(TilerError):
+    """Backend error."""
+
+
 PGSTAC_STATUS_CODES = {
     ReadOnlyPgSTACError: status.HTTP_500_INTERNAL_SERVER_ERROR,
+    BackendError: status.HTTP_500_INTERNAL_SERVER_ERROR,
     NoLayerFound: status.HTTP_400_BAD_REQUEST,
     MosaicNotFoundError: status.HTTP_404_NOT_FOUND,
 }

titiler/pgstac/factory.py

@@ -24,7 +24,12 @@
 from titiler.mosaic.factory import MosaicTilerFactory as BaseFactory
 from titiler.pgstac import model
 from titiler.pgstac.backend import PGSTACBackend
-from titiler.pgstac.dependencies import BackendParams, PgSTACParams, SearchParams
+from titiler.pgstac.dependencies import (
+    BackendParams,
+    PgSTACParams,
+    SearchParams,
+    register_search_in_db,
+)
 from titiler.pgstac.errors import ReadOnlyPgSTACError
 from titiler.pgstac.reader import SimpleSTACReader
 
@@ -117,15 +122,7 @@ def register_search(request: Request, search_query=Depends(search_dependency)):
                     conn.rollback()
                     pass
 
-                cursor.row_factory = class_row(model.Search)
-                cursor.execute(
-                    "SELECT * FROM search_query(%s, _metadata => %s);",
-                    (
-                        search.model_dump_json(by_alias=True, exclude_none=True),
-                        metadata.model_dump_json(exclude_none=True),
-                    ),
-                )
-                search_info = cursor.fetchone()
+                search_info = register_search_in_db(cursor, search, metadata)
 
         links: list[model.Link] = []