-
-
Notifications
You must be signed in to change notification settings - Fork 105
/
main.py
102 lines (85 loc) · 3.2 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
import json
import traceback
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.requests import HTTPConnection
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import PlainTextResponse
from starlette.staticfiles import StaticFiles
from api import settings as settings_module
from api import utils
from api.constants import VERSION
from api.ext import tor as tor_ext
from api.logger import get_logger
from api.settings import Settings
from api.utils.logging import log_errors
from api.views import router
logger = get_logger(__name__)
class RawContextMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send) -> None:
if scope["type"] not in ("http", "websocket"):
await self.app(scope, receive, send)
return
request = HTTPConnection(scope, receive)
token = settings_module.settings_ctx.set(request.app.settings)
try:
await self.app(scope, receive, send)
finally:
settings_module.settings_ctx.reset(token)
def get_app():
settings = Settings()
@asynccontextmanager
async def lifespan(app: FastAPI):
app.ctx_token = settings_module.settings_ctx.set(app.settings) # for events context
await settings.init()
await settings.plugins.startup()
yield
await app.settings.shutdown()
await settings.plugins.shutdown()
settings_module.settings_ctx.reset(app.ctx_token)
app = FastAPI(
title=settings.api_title,
version=VERSION,
docs_url="/",
redoc_url="/redoc",
root_path=settings.root_path,
description="Bitcart Merchants API",
lifespan=lifespan,
)
app.settings = settings
app.mount("/images", StaticFiles(directory=settings.images_dir), name="images")
app.mount("/files/localstorage", StaticFiles(directory=settings.files_dir), name="files")
app.include_router(router)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["Content-Disposition"],
)
settings.init_logging()
settings.load_plugins()
settings.plugins.setup_app(app)
@app.middleware("http")
async def add_onion_host(request: Request, call_next):
response = await call_next(request)
async with utils.redis.wait_for_redis():
host = request.headers.get("host", "").split(":")[0]
onion_host = await tor_ext.get_data("onion_host", "")
if onion_host and not tor_ext.is_onion(host):
response.headers["Onion-Location"] = onion_host + request.url.path
return response
@app.exception_handler(500)
async def exception_handler(request, exc):
logger.error(traceback.format_exc())
return PlainTextResponse("Internal Server Error", status_code=500)
app.add_middleware(RawContextMiddleware)
if settings.openapi_path:
with log_errors():
with open(settings.openapi_path) as f:
app.openapi_schema = json.loads(f.read())
return app
app = get_app()