xzbautomarkapi/core/server.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from fastapi import FastAPI, Request, Response
from starlette.middleware.cors import CORSMiddleware

from admin.api.router import router as admin_router
from app.api.router import router as app_router
from auth import router as auth_router
from common.endpoints import router as common_router
from core.config import settings


def create_app():

    app = FastAPI(title=settings.PROJECT_NAME,
                  openapi_url=f"/{settings.API_V1_STR}/openapi.json")

    # cors
    register_cors(app)

    # router
    register_router(app)

    # exception
    register_exception(app)

    # interceptor
    register_interceptor(app)

    # middlewares
    # app.add_middleware(CasbinMiddleware, enforcer=enforcer)
    # app.add_middleware(AuthenticationMiddleware, backend=JwtAuth())
    return app


def register_cors(app: FastAPI) -> None:
    if settings.DEBUG and settings.BACKEND_CORS_ORIGINS:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=[
                str(origin) for origin in settings.BACKEND_CORS_ORIGINS
            ],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )


def register_router(app: FastAPI) -> None:
    routers = [admin_router, auth_router, app_router, common_router]
    for router in routers:
        app.include_router(router, prefix="/" + settings.API_V1_STR)


def register_exception(app: FastAPI) -> None:
    pass


def register_interceptor(app: FastAPI) -> None:
    """
    请求响应拦截器
    """

    @app.middleware("http")
    async def logger_request(request: Request, call_next) -> Response:
        print(
            f"Method:{request.method}, Url:{request.url}, IP:{request.client.host}"
        )
        return await call_next(request)