-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathmain.py
167 lines (127 loc) · 6.99 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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
import logging
import logging.config
import os
from hdx_hapi.endpoints.exception_handler.request_validation_error_handler import request_validation_error_handler
from hdx_hapi.endpoints.util.exceptions import RequestParamsValidationError
logging.config.fileConfig(os.getenv('LOGGING_CONF_FILE', 'logging.conf'))
import uvicorn # noqa
from fastapi import FastAPI, Request # noqa
from fastapi.exceptions import ResponseValidationError # noqa
from fastapi.responses import HTMLResponse, RedirectResponse # noqa
from fastapi.openapi.docs import get_swagger_ui_html # noqa
# from hdx_hapi.services.sql_alchemy_session import init_db
from hdx_hapi.endpoints.exception_handler.response_validation_error_handler import response_validation_error_handler # noqa
from hdx_hapi.endpoints.middleware.app_identifier_middleware import app_identifier_middleware # noqa
from hdx_hapi.endpoints.middleware.mixpanel_tracking_middleware import mixpanel_tracking_middleware # noqa
from hdx_hapi.endpoints.get_encoded_identifier import router as encoded_identifier_router # noqa
from hdx_hapi.endpoints.get_request_verification import router as request_verification_router # noqa
from hdx_hapi.endpoints.favicon import router as favicon_router # noqa
from hdx_hapi.endpoints.get_population import router as population_router # noqa
from hdx_hapi.endpoints.get_operational_presence import router as operational_presence_router # noqa
from hdx_hapi.endpoints.get_funding import router as funding_router # noqa
from hdx_hapi.endpoints.get_conflict_events import router as conflict_events_router # noqa
from hdx_hapi.endpoints.get_admin_level import router as admin_level_router # noqa
from hdx_hapi.endpoints.get_hdx_metadata import router as dataset_router # noqa
from hdx_hapi.endpoints.get_humanitarian_response import router as humanitarian_response_router # noqa
from hdx_hapi.endpoints.get_affected_people import router as affected_people_router # noqa
from hdx_hapi.endpoints.get_national_risk import router as national_risk_router # noqa
from hdx_hapi.endpoints.get_wfp_commodity import router as wfp_commodity_router # noqa
from hdx_hapi.endpoints.get_wfp_market import router as wfp_market_router # noqa
from hdx_hapi.endpoints.get_currency import router as currency_router # noqa
from hdx_hapi.endpoints.get_food_security import router as food_security_router # noqa
from hdx_hapi.endpoints.get_food_price import router as food_price_router # noqa
from hdx_hapi.endpoints.get_data_availability import router as data_availability_router # noqa
from hdx_hapi.endpoints.get_idps import router as idps_router # noqa
from hdx_hapi.endpoints.get_returnees import router as returnees_router # noqa
from hdx_hapi.endpoints.get_version import router as version_router # noqa
from hdx_hapi.endpoints.util.version import api_version # noqa
# from hdx_hapi.endpoints.delete_example import delete_dataset
from hdx_hapi.config.config import get_config # noqa
logger = logging.getLogger(__name__)
CONFIG = get_config()
DESCRIPTION = """
The Humanitarian API (HDX HAPI) is a service of the
<a href="https://data.humdata.org">Humanitarian Data Exchange (HDX)</a>, part of OCHA\'s
<a href="https://centre.humdata.org">Centre for Humanitarian Data</a>.
\nThis is the reference documentation of the API.
You may want to <a href="https://hdx-hapi.readthedocs.io/en/latest/">get started here.</a>
All queries require an `app_identifier` which can be supplied as a query parameter or as a header
(`X-HDX-HAPI-APP-IDENTIFIER`). The `app_identifier` is simply a base64 encoded version of a user supplied
application name and email address.
The `limit` and `offset` parameters are available for all queries and have the usual database meanings
to provide pagination of results. If no `limit` is specified, a maximum of 10,000 records will be returned.
The `output_format` parameter is available for all queries and can be set to JSON or csv,
where JSON is selected rows of data are supplied under a data key.
Query parameters that access string fields are implicitly wildcards and case insensitive
so that `location_name=Mali` will return data for Mali and Somalia.
""" # noqa
app = FastAPI(
title='HDX HAPI',
description=DESCRIPTION,
version=api_version,
docs_url=None,
servers=[{'url': CONFIG.HAPI_SERVER_URL}] if CONFIG.HAPI_SERVER_URL else [],
)
app.include_router(encoded_identifier_router)
app.include_router(request_verification_router)
app.include_router(favicon_router)
app.include_router(affected_people_router)
app.include_router(idps_router)
app.include_router(returnees_router)
app.include_router(operational_presence_router)
app.include_router(funding_router)
app.include_router(conflict_events_router)
app.include_router(national_risk_router)
app.include_router(food_security_router)
app.include_router(food_price_router)
app.include_router(population_router)
app.include_router(dataset_router)
app.include_router(admin_level_router)
app.include_router(currency_router)
app.include_router(humanitarian_response_router)
app.include_router(wfp_commodity_router)
app.include_router(wfp_market_router)
app.include_router(data_availability_router)
app.include_router(version_router)
# add middleware
@app.middleware('http')
async def app_identifier_middleware_init(request: Request, call_next):
response = await app_identifier_middleware(request, call_next)
return response
# add middleware
@app.middleware('http')
async def mixpanel_tracking_middleware_init(request: Request, call_next):
response = await mixpanel_tracking_middleware(request, call_next)
return response
@app.on_event('startup')
async def startup():
# await init_db()
# await populate_db()
pass
# Adding custom favicon based on article
@app.get('/docs', include_in_schema=False)
async def swagger_ui_html(req: Request) -> HTMLResponse:
root_path = req.scope.get('root_path', '').rstrip('/')
openapi_url = root_path + app.openapi_url # pyright: ignore[reportOperatorIssue]
oauth2_redirect_url = app.swagger_ui_oauth2_redirect_url
if oauth2_redirect_url:
oauth2_redirect_url = root_path + oauth2_redirect_url
return get_swagger_ui_html(
openapi_url=openapi_url,
title=app.title + ' - OpenAPI Docs',
oauth2_redirect_url=oauth2_redirect_url,
init_oauth=app.swagger_ui_init_oauth,
swagger_favicon_url='/favicon.ico',
swagger_ui_parameters=app.swagger_ui_parameters,
)
@app.get('/', include_in_schema=False)
def home():
return RedirectResponse('/docs')
@app.exception_handler(ResponseValidationError)
async def resp_validation_exception_handler(request: Request, exc: ResponseValidationError):
return await response_validation_error_handler(request, exc)
@app.exception_handler(RequestParamsValidationError)
async def req_params_validation_exception_handler(request: Request, exc: RequestParamsValidationError):
return await request_validation_error_handler(request, exc)
if __name__ == '__main__':
uvicorn.run(app, host='0.0.0.0', port=8844, log_config=os.getenv('LOGGING_CONF_FILE', 'logging.conf'))