python-base--/fastapi-demo/main.py

import importlib
import json
import os
import time

import fastapi
from fastapi import FastAPI
from fastapi import HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel  # 参数校验
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import JSONResponse, RedirectResponse
from starlette.staticfiles import StaticFiles

from config import logger, CfgBaseInit

app = FastAPI(title="project name ", description="通用系统 ", version="v 0.0.0")

# app.routes.pop(1)  # drop default docs_url router

# 添加 session 中间键，使项目中可以使用session
app.add_middleware(SessionMiddleware, secret_key='123456hhh')
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*", ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

static_obj = StaticFiles(directory="static", html=True)
app.mount("/static", static_obj, "static")


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    """接口服务异常捕获; text/plain"""
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """请求校验异常捕获; application/json """
    json_response = JSONResponse({'message': '【 {} 】接口触发了 RequestValidationError 错误.'
                                 .format(request.scope.get("endpoint").__name__, exc.errors()), 'status_code': 999,
                                  "error_message": exc.errors(), "queryParasm": exc.body})
    logger.error(json.loads(json_response.body))
    return json_response


@app.exception_handler(Exception)
async def validation_exception_handler(request, exc):
    """请求校验异常捕获; application/json """
    return JSONResponse({'message': "服务器内部错误", 'status_code': 500})


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """接口响应中间键; 当前只支持 http 请求"""
    logger.info(f'【 {request.scope.get("path")} 】API query start...')
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["API-Process-Time"] = f"{process_time} seconds"
    logger.info(f'【 {request.scope.get("path")} 】API, Time:【 {process_time} seconds 】...')
    return response


# 动态注册路由
for file in os.listdir(os.path.join(CfgBaseInit.project_path, "controller")):
    if not file.endswith("py") or '__init__.py' in file:
        continue
    file = file.split('.')[0]
    router_module = importlib.import_module(f"controller.{file}", package=None)
    if hasattr(router_module, "router"):
        # 注册路由
        app.include_router(
            router_module.router,
            # dependencies=[Depends(get_token_header)],
            responses={404: {"description": "Not found"}},
        )


@app.get("/")
def index():
    return RedirectResponse(url="/static/index.html")


# http://localhost:8000/query?uid=1
@app.get("/query")
def query(uid):
    msg = f'uid为{uid}'
    return {'success': True, 'msg': msg}


# http://localhost:8000/query/1   参数类型有校验
@app.get("/query/{uid}")
def query(uid: int):
    msg = f'uid为{uid}'
    return {'success': True, 'msg': msg}


class People(BaseModel):
    name: str
    age: int
    address: str
    salary: float


# json body请求
@app.post("/insert")
def insert(people: People):
    msg = f"名字：{people.name},年龄{people.age}"
    return msg