-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
117 lines (87 loc) · 3.11 KB
/
app.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
import json
import time
from typing import Any
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
from routers.middlewares.request_id import RequestIDContextMiddleware
from routers.validator import router as validator_router
from config import Config
from utils.setup_logger import setup_logging
app: FastAPI = FastAPI(
title='The Validator by Zensum',
description=f"""
A __Validator service__ built at Zensum.
---
### Code
You can find this code [here on Github](https://github.com/zensum/validator-service).
---
### Features
- Validating for Sweden and Norway
- emails
- phone numbers
- account numbers
- personal ids
---
### Request ID
Please add a value to the header `{Config.request_id_key}` with a UUID4 value to be able to track
the request between servies. 🙏🏼
---
### Support
Pray 🙏🏼
---
""",
openapi_tags=[
{
'name': 'Validator',
'description': '__Validating__. This method validates emails, phone numbers, pni and accounts. '
},
],
)
if Config.REQUIRE_HTTPS:
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)
@app.on_event('startup')
async def startup_event() -> None:
setup_logging()
# noinspection PyStatementEffect
# Checks that all necessary envs are set with an overkill map
*map(lambda k: k.startswith('__') or getattr(Config, k), dir(Config.__class__)),
@app.middleware('http')
async def add_process_time_header(request: Request, call_next): # type: ignore
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers['X-Process-Time'] = str(round(process_time * 1000)) + ' ms'
if response.headers.get('server'):
del response.headers['server']
return response
async def http_exception_middleware(request: Request, call_next: Any) -> Response:
try:
return await call_next(request)
except HTTPException as e:
return Response(json.dumps(dict(detail=e.detail)), status_code=e.status_code, headers=e.headers)
app.add_middleware(RequestIDContextMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=Config.Origins,
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'],
expose_headers=[Config.request_id_key, 'Content-Type'],
)
@app.middleware('http')
async def add_cache_header(request: Request, call_next): # type: ignore
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers['X-Process-Time'] = str(round(process_time * 1000)) + ' ms'
response.headers['Cache-Control'] = f'max-age={10*60}' # 10 min
return response
app.include_router(
validator_router,
tags=['Validator'],
)
# this must be last middleware registered to catch all HTTPException exceptions in middlewares and convert to response
app.add_middleware(BaseHTTPMiddleware, dispatch=http_exception_middleware)