Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/cicd.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ jobs:
if: ${{ matrix.python-version == env.LATEST_PY_VERSION }}
run: |
uv run pre-commit run --all-files
uv run --with mypy --with types-attrs mypy -p stac_fastapi

- name: install lib postgres
uses: nyurik/action-setup-postgis@v2
Expand Down
2 changes: 2 additions & 0 deletions CHANGES.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
### Fixed

- Update pydantic and pydantic-settings versions requirements
- Improve type hints


## [6.1.4] - 2025-12-08

Expand Down
33 changes: 20 additions & 13 deletions stac_fastapi/pgstac/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

import os
from contextlib import asynccontextmanager
from typing import Dict, List, Set, Type, cast

from brotli_asgi import BrotliMiddleware
from fastapi import APIRouter, FastAPI
Expand Down Expand Up @@ -36,6 +37,8 @@
from stac_fastapi.extensions.core.query import QueryConformanceClasses
from stac_fastapi.extensions.core.sort import SortConformanceClasses
from stac_fastapi.extensions.third_party import BulkTransactionExtension
from stac_fastapi.types.extension import ApiExtension
from stac_fastapi.types.search import APIRequest
from starlette.middleware import Middleware

from stac_fastapi.pgstac.config import Settings
Expand All @@ -49,7 +52,7 @@
settings = Settings()

# search extensions
search_extensions_map = {
search_extensions_map: Dict[str, ApiExtension] = {
"query": QueryExtension(),
"sort": SortExtension(),
"fields": FieldsExtension(),
Expand All @@ -58,7 +61,7 @@
}

# collection_search extensions
cs_extensions_map = {
cs_extensions_map: Dict[str, ApiExtension] = {
"query": QueryExtension(conformance_classes=[QueryConformanceClasses.COLLECTIONS]),
"sort": SortExtension(conformance_classes=[SortConformanceClasses.COLLECTIONS]),
"fields": FieldsExtension(conformance_classes=[FieldsConformanceClasses.COLLECTIONS]),
Expand All @@ -70,7 +73,7 @@
}

# item_collection extensions
itm_col_extensions_map = {
itm_col_extensions_map: Dict[str, ApiExtension] = {
"query": QueryExtension(
conformance_classes=[QueryConformanceClasses.ITEMS],
),
Expand All @@ -82,7 +85,7 @@
"pagination": TokenPaginationExtension(),
}

enabled_extensions = {
enabled_extensions: Set[str] = {
*search_extensions_map.keys(),
*cs_extensions_map.keys(),
*itm_col_extensions_map.keys(),
Expand All @@ -92,7 +95,7 @@
if ext := os.environ.get("ENABLED_EXTENSIONS"):
enabled_extensions = set(ext.split(","))

application_extensions = []
application_extensions: List[ApiExtension] = []

with_transactions = os.environ.get("ENABLE_TRANSACTIONS_EXTENSIONS", "").lower() in [
"yes",
Expand Down Expand Up @@ -123,23 +126,27 @@
application_extensions.extend(search_extensions)

# /collections/{collectionId}/items model
items_get_request_model = ItemCollectionUri
items_get_request_model: Type[APIRequest] = ItemCollectionUri
itm_col_extensions = [
extension
for key, extension in itm_col_extensions_map.items()
if key in enabled_extensions
]
if itm_col_extensions:
items_get_request_model = create_request_model(
model_name="ItemCollectionUri",
base_model=ItemCollectionUri,
extensions=itm_col_extensions,
request_type="GET",
items_get_request_model = cast(
Type[APIRequest],
create_request_model(
model_name="ItemCollectionUri",
base_model=ItemCollectionUri,
extensions=itm_col_extensions,
request_type="GET",
),
)

application_extensions.extend(itm_col_extensions)

# /collections model
collections_get_request_model = EmptyRequest
collections_get_request_model: Type[APIRequest] = EmptyRequest
if "collection_search" in enabled_extensions:
cs_extensions = [
extension
Expand Down Expand Up @@ -191,7 +198,7 @@ async def lifespan(app: FastAPI):
allow_headers=settings.cors_headers,
),
],
health_check=health_check,
health_check=health_check, # type: ignore [arg-type]
)
app = api.app

Expand Down
Loading
Loading