72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244 | @typing.no_type_check
def init_app( # pylint: disable=too-many-locals, too-many-statements
api_version: str,
routers: list[APIRouter],
router_prefix: str = "",
) -> FastAPI: # pylint: disable=too-many-arguments
"""
Init the FastAPI application.
See: https://praciano.com.br/fastapi-and-async-sqlalchemy-20-with-pytest-done-right.html
Args:
api_version (str): version of our application (not the version of the OpenAPI specification
nor the version of FastAPI being used)
routers (list[APIRouter]): list of FastAPI routers to add to the application.
router_prefix (str): used by stac_fastapi
"""
@asynccontextmanager
async def lifespan(*_):
"""Automatically executed when starting and stopping the FastAPI server."""
###########
# STARTUP #
###########
# Init objects for dependency injection
settings.set_http_client(httpx.AsyncClient(timeout=DEFAULT_TIMEOUT_CONFIG))
yield
############
# SHUTDOWN #
############
# Close objects for dependency injection
await settings.del_http_client()
# Init the FastAPI application
app = FastAPI(title="RS-Server", version=api_version, lifespan=lifespan, **docs_params(router_prefix))
# Configure OpenTelemetry
init_opentelemetry.init_traces(app, settings.SERVICE_NAME)
# Init a pgstac client for adgs and cadip.
# TODO: remove this when adgs and cadip switch to a stac_fastapi application.
# Example taken from: https://github.com/stac-utils/stac-fastapi-pgstac/blob/main/tests/api/test_api.py
app.state.router_prefix = router_prefix # NOTE: maybe we should keep this one
extensions = [ # no transactions because we don't update the database
# TransactionExtension(client=TransactionsClient(), settings=api_settings),
QueryExtension(),
SortExtension(),
FieldsExtension(),
FilterExtension(client=FiltersClient()),
PaginationExtension(),
# BulkTransactionExtension(client=BulkTransactionsClient()),
]
search_post_request_model = create_post_request_model(extensions, base_model=PgstacSearch)
app.state.pgstac_client = CoreCrudClient(pgstac_search_model=search_post_request_model)
# patch the pgstac_client.landing_page method to add "rel": "child" link for each collection
async def patched_landing_page(self, request, **kwargs):
# Call the original method
original = await CoreCrudClient.landing_page(self, request=request, **kwargs)
# Get base from 'self' link
base = next((link["href"] for link in original["links"] if link.get("rel") == "self"), "").rstrip("/") + "/"
# Fetch collections
collections = (await self.all_collections(request=request)).get("collections", [])
# Append rel="child" links
original["links"] += [
{
"rel": "child",
"type": "application/json",
"title": collection.get("title") or collection["id"],
"href": urljoin(base, f"collections/{collection['id']}"),
}
for collection in collections
]
return original
# Monkey patch the pgstac_client.landing_page method
app.state.pgstac_client.landing_page = MethodType(patched_landing_page, app.state.pgstac_client)
# TODO: remove this when adgs and cadip switch to a stac_fastapi application.
app.state.pgstac_client.extensions = extensions
for ext in extensions:
ext.register(app)
app.state.pgstac_client.title = app.title
app.state.pgstac_client.description = app.description
# Implement the /search endpoints by simulating a StacApi object, TODO remove this also
app.settings = State()
app.settings.enable_response_models = False
app.settings.use_api_hydrate = False
app.state.settings = app.settings
app.client = app.state.pgstac_client
app.search_get_request_model = create_get_request_model(extensions)
app.search_post_request_model = search_post_request_model
app.router.prefix = router_prefix # TODO should be used by other endpoints ?
StacApi.register_get_search(app)
StacApi.register_post_search(app)
app.router.prefix = ""
if settings.CLUSTER_MODE:
scopes = [] # One scope for each Router path and method
for route in app.router.routes:
if not isinstance(route, APIRoute):
continue
for method_ in route.methods:
scopes.append({"path": route.path, "method": method_})
add_route_dependencies(app.router.routes, scopes=scopes, dependencies=[Depends(authenticate)])
# TODO: title and description must be set using the env vars
# CATALOG_METADATA_TITLE and CATALOG_METADATA_DESCRIPTION
service = router_prefix.strip("/").title()
app.state.pgstac_client.title = f"RS-PYTHON {service} collections"
app.state.pgstac_client.description = f"{service} collections of Copernicus Reference System Python"
# By default FastAPI will return 422 status codes for invalid requests
# But the STAC api spec suggests returning a 400 in this case
# TODO remove this also
add_exception_handlers(app, {})
dependencies = []
if settings.CLUSTER_MODE:
# Get the oauth2 router
oauth2_router = oauth2.get_router(app)
# Add it to the FastAPI application
app.include_router(
oauth2_router,
tags=["Authentication"],
prefix=AUTH_PREFIX,
include_in_schema=True,
)
# Add the api key / oauth2 security: the user must provide
# an api key (generated from the apikey manager) or authenticate to the
# oauth2 service (keycloak) to access the endpoints
dependencies.append(Depends(authenticate))
# Add all the input routers (and not the oauth2 nor technical routers) to a single bigger router
# to which we add the authentication dependency.
need_auth_router = APIRouter(dependencies=dependencies)
for router in routers:
need_auth_router.include_router(router)
# Add routers to the FastAPI app
app.include_router(need_auth_router)
app.include_router(technical_router)
# Catch all exceptions and return a JSONResponse
app.add_middleware(HandleExceptionsMiddleware)
# This middleware allows to have consistant http/https protocol in stac links
app.add_middleware(ProxyHeaderMiddleware)
# Middleware for implementing first and last buttons in STAC Browser
app.add_middleware(PaginationLinksMiddleware)
app.add_middleware(StacLinksTitleMiddleware, title="My STAC Title")
# Add CORS requests from the STAC browser
if settings.CORS_ORIGINS:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
)
# Finally, apply stac-fastapi openapi patch to comply with the STAC API spec
return update_openapi(app)
|
