import importlib import json import os import time 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 from starlette.staticfiles import StaticFiles from config import logger, CfgBaseInit app = FastAPI(title="EIP REC ENGINE ", description="通用系统 ", version="v 0.0.0") app.routes.pop(1) # drop default docs_url router # 添加 session 中间键,使项目中可以使用session app.add_middleware(SessionMiddleware, secret_key='123456') # 跨域白名单, 解决跨域问题 origins = [ "*", ] app.add_middleware( CORSMiddleware, allow_origins=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").replace("/", ".").lstrip(".")} 】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" response.headers["service"] = "EIP REC ENGINE" response.headers["Request-Client"] = str(request.scope.get("client")[0]) + ":" + str( request.scope.get("client")[1]) response.headers["User-Agent"] = str(request.headers.get("user-agent")) logger.info(f'Request-Client is :【 {response.headers.get("Request-Client")} 】') logger.info(f'Request-Client User-Agent is :【 {response.headers.get("User-Agent")} 】') request.scope.get("path").replace("/", "") logger.info(f'【 {request.scope.get("path")} 】API ' f'Process-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 {"message": "你已经正确创建api"} # 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 # jsonbody请求 @app.post("/insert") def insert(people: People): msg = f"名字:{people.name},年龄{people.age}" return msg