Spaces:
Sleeping
Sleeping
File size: 3,505 Bytes
5564ecb |
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 |
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from asgi_correlation_id import CorrelationIdMiddleware
from fastapi import Depends, FastAPI
from fastapi_limiter import FastAPILimiter
from fastapi_pagination import add_pagination
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.staticfiles import StaticFiles
from common.exception.exception_handler import register_exception
from common.log import set_custom_logfile, setup_logging
from core.conf import settings
from core.path_conf import STATIC_DIR, UPLOAD_DIR
from utils.health_check import ensure_unique_route_names, http_limit_callback
from utils.openapi import simplify_operation_ids
from utils.serializers import MsgSpecJSONResponse
# @asynccontextmanager
# async def register_init(app: FastAPI) -> AsyncGenerator[None, None]:
# """
# :param app: FastAPI
# :return:
# """
# await create_table()
# # 连接 redis
# await redis_client.open()
# await FastAPILimiter.init(
# redis=redis_client,
# prefix=settings.REQUEST_LIMITER_REDIS_PREFIX,
# http_callback=http_limit_callback,
# )
# yield
# await redis_client.close()
# await FastAPILimiter.close()
def register_app() -> FastAPI:
""" FastAPI """
app = FastAPI(
title=settings.FASTAPI_TITLE,
version=settings.FASTAPI_VERSION,
description=settings.FASTAPI_DESCRIPTION,
docs_url=settings.FASTAPI_DOCS_URL,
redoc_url=settings.FASTAPI_REDOC_URL,
openapi_url=settings.FASTAPI_OPENAPI_URL,
default_response_class=MsgSpecJSONResponse,
)
register_logger()
register_static_file(app)
register_middleware(app)
register_router(app)
register_page(app)
register_exception(app)
return app
def register_logger() -> None:
setup_logging()
set_custom_logfile()
def register_static_file(app: FastAPI) -> None:
"""
:param app: FastAPI
:return:
"""
if not os.path.exists(UPLOAD_DIR):
os.makedirs(UPLOAD_DIR)
app.mount('/static/upload', StaticFiles(directory=UPLOAD_DIR), name='upload')
if settings.FASTAPI_STATIC_FILES:
app.mount('/static', StaticFiles(directory=STATIC_DIR), name='static')
def register_middleware(app: FastAPI) -> None:
"""
:param app: FastAPI
:return:
"""
# Opera log
# app.add_middleware(OperaLogMiddleware)
# # Access log
# if settings.MIDDLEWARE_ACCESS:
# from middleware.access_middleware import AccessMiddleware
# app.add_middleware(AccessMiddleware)
# Trace ID
app.add_middleware(CorrelationIdMiddleware, validator=False)
# CORS
if settings.MIDDLEWARE_CORS:
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'],
expose_headers=settings.CORS_EXPOSE_HEADERS,
)
def register_router(app: FastAPI) -> None:
"""
:param app: FastAPI
:return:
"""
from app.router import router
app.include_router(router)
ensure_unique_route_names(app)
simplify_operation_ids(app)
def register_page(app: FastAPI) -> None:
"""
:param app: FastAPI
:return:
"""
add_pagination(app)
|