-
Notifications
You must be signed in to change notification settings - Fork 24
/
main.py
executable file
·151 lines (116 loc) · 4.11 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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
#!/usr/bin/env python3
import logging
import os
from typing import Awaitable, Callable, Dict, List
from fastapi import FastAPI, Request, Response
from fastapi.openapi.utils import get_openapi
from morecantile.defaults import tms as defaultTileMatrices
from morecantile.models import TileMatrixSet
from starlette.middleware.cors import CORSMiddleware
from titiler.core.errors import DEFAULT_STATUS_CODES, add_exception_handlers
from titiler.core.middleware import (
CacheControlMiddleware,
LoggerMiddleware,
TotalTimeMiddleware,
)
from titiler.mosaic.errors import MOSAIC_STATUS_CODES
from titiler.pgstac.db import close_db_connection, connect_to_db
from pccommon.constants import X_REQUEST_ENTITY
from pccommon.logging import ServiceName, init_logging
from pccommon.middleware import (
RequestTracingMiddleware,
handle_exceptions,
timeout_middleware,
)
from pccommon.openapi import fixup_schema
from pctiler.config import get_settings
from pctiler.endpoints import health, item, legend, pg_mosaic
# Initialize logging
init_logging(ServiceName.TILER)
logger = logging.getLogger(__name__)
# Get the root path if set in the environment
APP_ROOT_PATH = os.environ.get("APP_ROOT_PATH", "")
settings = get_settings()
app = FastAPI(
title=settings.title,
openapi_url=settings.openapi_url,
root_path=APP_ROOT_PATH,
)
app.state.service_name = ServiceName.TILER
app.include_router(
item.pc_tile_factory.router,
prefix=settings.item_endpoint_prefix,
tags=["Item tile endpoints"],
)
app.include_router(
pg_mosaic.pgstac_mosaic_factory.router,
prefix=settings.mosaic_endpoint_prefix,
tags=["PgSTAC Mosaic endpoints"],
)
app.include_router(
legend.legend_router,
prefix=settings.legend_endpoint_prefix,
tags=["Legend endpoints"],
)
app.include_router(health.health_router, tags=["Liveliness/Readiness"])
app.add_middleware(RequestTracingMiddleware, service_name=ServiceName.TILER)
@app.middleware("http")
async def _timeout_middleware(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
"""Add a timeout to all requests."""
return await timeout_middleware(request, call_next, timeout=settings.request_timout)
@app.middleware("http")
async def _handle_exceptions(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
return await handle_exceptions(request, call_next)
add_exception_handlers(app, DEFAULT_STATUS_CODES)
add_exception_handlers(app, MOSAIC_STATUS_CODES)
app.add_middleware(CacheControlMiddleware, cachecontrol="public, max-age=3600")
app.add_middleware(TotalTimeMiddleware)
if settings.debug:
app.add_middleware(LoggerMiddleware)
# Note: If requests are being sent through an application gateway like
# nginx-ingress, you may need to configure CORS through that system.
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST"],
allow_headers=[X_REQUEST_ENTITY],
)
@app.on_event("startup")
async def startup_event() -> None:
"""Connect to database on startup."""
await connect_to_db(app)
@app.on_event("shutdown")
async def shutdown_event() -> None:
"""Close database connection."""
await close_db_connection(app)
@app.get("/")
async def read_root() -> Dict[str, str]:
return {"Hello": "Planetary Developer!"}
@app.get("/tileMatrixSets")
async def matrix_list() -> List[str]:
return defaultTileMatrices.list()
@app.get("/tileMatrixSets/{tileMatrixSetId}")
async def matrix_definition(tileMatrixSetId: str) -> TileMatrixSet:
logger.info(
"Matrix definition requested",
extra={
"custom_dimensions": {
"tileMatrixSetId": tileMatrixSetId,
}
},
)
return defaultTileMatrices.get(tileMatrixSetId)
def custom_openapi() -> Dict:
if not app.openapi_schema:
schema = get_openapi(
title="Preview of Tile Access Features",
version=settings.api_version,
routes=app.routes,
)
app.openapi_schema = fixup_schema(app.root_path, schema)
return app.openapi_schema # type: ignore
app.openapi = custom_openapi # type: ignore