77
88import os
99from contextlib import asynccontextmanager
10+ from typing import Dict , List , Set , Type , cast
1011
1112from brotli_asgi import BrotliMiddleware
1213from fastapi import APIRouter , FastAPI
3637from stac_fastapi .extensions .core .query import QueryConformanceClasses
3738from stac_fastapi .extensions .core .sort import SortConformanceClasses
3839from stac_fastapi .extensions .third_party import BulkTransactionExtension
40+ from stac_fastapi .types .extension import ApiExtension
41+ from stac_fastapi .types .search import APIRequest
3942from starlette .middleware import Middleware
4043
4144from stac_fastapi .pgstac .config import Settings
4952settings = Settings ()
5053
5154# search extensions
52- search_extensions_map = {
55+ search_extensions_map : Dict [ str , ApiExtension ] = {
5356 "query" : QueryExtension (),
5457 "sort" : SortExtension (),
5558 "fields" : FieldsExtension (),
5861}
5962
6063# collection_search extensions
61- cs_extensions_map = {
64+ cs_extensions_map : Dict [ str , ApiExtension ] = {
6265 "query" : QueryExtension (conformance_classes = [QueryConformanceClasses .COLLECTIONS ]),
6366 "sort" : SortExtension (conformance_classes = [SortConformanceClasses .COLLECTIONS ]),
6467 "fields" : FieldsExtension (conformance_classes = [FieldsConformanceClasses .COLLECTIONS ]),
7073}
7174
7275# item_collection extensions
73- itm_col_extensions_map = {
76+ itm_col_extensions_map : Dict [ str , ApiExtension ] = {
7477 "query" : QueryExtension (
7578 conformance_classes = [QueryConformanceClasses .ITEMS ],
7679 ),
8285 "pagination" : TokenPaginationExtension (),
8386}
8487
85- enabled_extensions = {
88+ enabled_extensions : Set [ str ] = {
8689 * search_extensions_map .keys (),
8790 * cs_extensions_map .keys (),
8891 * itm_col_extensions_map .keys (),
9295if ext := os .environ .get ("ENABLED_EXTENSIONS" ):
9396 enabled_extensions = set (ext .split ("," ))
9497
95- application_extensions = []
98+ application_extensions : List [ ApiExtension ] = []
9699
97100with_transactions = os .environ .get ("ENABLE_TRANSACTIONS_EXTENSIONS" , "" ).lower () in [
98101 "yes" ,
123126application_extensions .extend (search_extensions )
124127
125128# /collections/{collectionId}/items model
126- items_get_request_model = ItemCollectionUri
129+ items_get_request_model : Type [ APIRequest ] = ItemCollectionUri
127130itm_col_extensions = [
128131 extension
129132 for key , extension in itm_col_extensions_map .items ()
130133 if key in enabled_extensions
131134]
132135if itm_col_extensions :
133- items_get_request_model = create_request_model (
134- model_name = "ItemCollectionUri" ,
135- base_model = ItemCollectionUri ,
136- extensions = itm_col_extensions ,
137- request_type = "GET" ,
136+ items_get_request_model = cast (
137+ Type [APIRequest ],
138+ create_request_model (
139+ model_name = "ItemCollectionUri" ,
140+ base_model = ItemCollectionUri ,
141+ extensions = itm_col_extensions ,
142+ request_type = "GET" ,
143+ ),
138144 )
145+
139146 application_extensions .extend (itm_col_extensions )
140147
141148# /collections model
142- collections_get_request_model = EmptyRequest
149+ collections_get_request_model : Type [ APIRequest ] = EmptyRequest
143150if "collection_search" in enabled_extensions :
144151 cs_extensions = [
145152 extension
@@ -173,7 +180,7 @@ async def lifespan(app: FastAPI):
173180 router = APIRouter (prefix = settings .prefix_path ),
174181 settings = settings ,
175182 extensions = application_extensions ,
176- client = CoreCrudClient (pgstac_search_model = post_request_model ),
183+ client = CoreCrudClient (pgstac_search_model = post_request_model ), # type: ignore [arg-type]
177184 response_class = JSONResponse ,
178185 items_get_request_model = items_get_request_model ,
179186 search_get_request_model = get_request_model ,
@@ -191,7 +198,7 @@ async def lifespan(app: FastAPI):
191198 allow_headers = settings .cors_headers ,
192199 ),
193200 ],
194- health_check = health_check ,
201+ health_check = health_check , # type: ignore [arg-type]
195202)
196203app = api .app
197204
0 commit comments