123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130 |
- 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
|