Make API customization easier

[mcucchi9]

Description

As an enthusiast user of the stac-fastapi-pgstac, I find it a little cumbersome to implement little customizations on top of it. I came to believe that this mainly stems from the structure of the app.py module, and that customizability could be improved by implementing a small (and mainly backward-compatible) refactoring of it.

Use cases

A couple of (real and personal) use cases which motivates this issue:

• adding a validation/post-processing step on top of the QueryExtension defined in the stac_fastapi.pgstac.extensions.query;
• patching the _search_base method of the stac_fastapi.pgstac.core.CoreCrudClient -> see Query with bbox crossing antimeridian does not work as expected pgstac#428 (comment);
• updating the model_config for stac_fastapi.pgstac.config.Settings.

Each one of the above, even taken independently, required copying a quite a few lines of boilerplate code from the app.py module.

Possible solution

One possible solution which I applied in another project of mine essentially relies on wrapping these lines

stac-fastapi-pgstac/stac_fastapi/pgstac/app.py

Lines 55 to 199 in df1de10

	search_extensions_map: dict[str, ApiExtension] = {
	"query": QueryExtension(),
	"sort": SortExtension(),
	"fields": FieldsExtension(),
	"filter": SearchFilterExtension(client=FiltersClient()),
	"pagination": TokenPaginationExtension(),
	}
	# collection_search extensions
	cs_extensions_map: dict[str, ApiExtension] = {
	"query": QueryExtension(conformance_classes=[QueryConformanceClasses.COLLECTIONS]),
	"sort": SortExtension(conformance_classes=[SortConformanceClasses.COLLECTIONS]),
	"fields": FieldsExtension(conformance_classes=[FieldsConformanceClasses.COLLECTIONS]),
	"filter": CollectionSearchFilterExtension(client=FiltersClient()),
	"free_text": FreeTextExtension(
	conformance_classes=[FreeTextConformanceClasses.COLLECTIONS],
	),
	"pagination": OffsetPaginationExtension(),
	}
	# item_collection extensions
	itm_col_extensions_map: dict[str, ApiExtension] = {
	"query": QueryExtension(
	conformance_classes=[QueryConformanceClasses.ITEMS],
	),
	"sort": SortExtension(
	conformance_classes=[SortConformanceClasses.ITEMS],
	),
	"fields": FieldsExtension(conformance_classes=[FieldsConformanceClasses.ITEMS]),
	"filter": ItemCollectionFilterExtension(client=FiltersClient()),
	"pagination": TokenPaginationExtension(),
	}
	enabled_extensions: set[str] = {
	*search_extensions_map.keys(),
	*cs_extensions_map.keys(),
	*itm_col_extensions_map.keys(),
	"collection_search",
	}
	if ext := settings.enabled_extensions:
	enabled_extensions = set(ext.split(","))
	application_extensions: list[ApiExtension] = []
	with_transactions = settings.enable_transactions_extensions
	if with_transactions:
	application_extensions.append(
	TransactionExtension(
	client=TransactionsClient(),
	settings=settings,
	response_class=JSONResponse,
	),
	)
	application_extensions.append(
	BulkTransactionExtension(client=BulkTransactionsClient()),
	)
	# /search models
	search_extensions = [
	extension
	for key, extension in search_extensions_map.items()
	if key in enabled_extensions
	]
	post_request_model = create_post_request_model(search_extensions, base_model=PgstacSearch)
	get_request_model = create_get_request_model(search_extensions)
	application_extensions.extend(search_extensions)
	# /collections/{collectionId}/items model
	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 = 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: type[APIRequest] = EmptyRequest
	if "collection_search" in enabled_extensions:
	cs_extensions = [
	extension
	for key, extension in cs_extensions_map.items()
	if key in enabled_extensions
	]
	collection_search_extension = CollectionSearchExtension.from_extensions(cs_extensions)
	collections_get_request_model = collection_search_extension.GET
	application_extensions.append(collection_search_extension)
	@asynccontextmanager
	async def lifespan(app: FastAPI):
	"""FastAPI Lifespan."""
	await connect_to_db(app, add_write_connection_pool=with_transactions)
	yield
	await close_db_connection(app)
	api = StacApi(
	app=FastAPI(
	openapi_url=settings.openapi_url,
	docs_url=settings.docs_url,
	redoc_url=None,
	root_path=settings.root_path,
	title=settings.stac_fastapi_title,
	version=settings.stac_fastapi_version,
	description=settings.stac_fastapi_description,
	lifespan=lifespan,
	),
	router=APIRouter(prefix=settings.prefix_path),
	settings=settings,
	extensions=application_extensions,
	client=CoreCrudClient(pgstac_search_model=post_request_model), # type: ignore [arg-type]
	response_class=JSONResponse,
	items_get_request_model=items_get_request_model,
	search_get_request_model=get_request_model,
	search_post_request_model=post_request_model,
	collections_get_request_model=collections_get_request_model,
	middlewares=[
	Middleware(BrotliMiddleware),
	Middleware(ProxyHeaderMiddleware),
	Middleware(
	CORSMiddleware,
	allow_origins=settings.cors_origins,
	allow_origin_regex=settings.cors_origin_regex,
	allow_methods=settings.cors_methods,
	allow_credentials=settings.cors_credentials,
	allow_headers=settings.cors_headers,
	max_age=600,
	),
	],
	health_check=health_check, # type: ignore [arg-type]
	)

into a function, such as instantiate_api, with arguments giving the possibility to inject e.g. a customized client and customized extensions.

The function defaults would leave current behavior unchanged. The only possible source of non-backward-compatibility I can foresee stems from the eventuality that some user currently imports objects from the app.py module's namespace which will then live inside the instantiate_api function.

I can work on a draft PR on this.

[jonhealy1]

This is a great idea and definitely long overdue.

[jonhealy1]

Feel free to add something similar to sfeos too - stac-fastapi-elasticsearch-opensearch

[mcucchi9]

@jonhealy1 and anyone else who might be interested, I drafted a first PR here: #381.

It is still a WIP (e.g. I still have to add tests for the customisation functionalities), but at least pre-existing tests pass. Please let me know if the general approach suits you, or if you any suggestion.