【FastAPI】 项目化结构

说明

现在网上关于 FastAPI 的项目结构的资料比较少，以下仅供参考，如有更好的想法，可私聊博主交流

项目下载：https://pan.bigdataboy.cn/#/s/oWsb

结构

说明：项目结构的静态文件使用 Jinja2 渲染

项目
├── modules （功能模块）
│ ├── 
│ └── init.py
├── routers   (路由控制)
│ ├── index.py
│ ├── items.py
│ ├── users.py
│ └── init.py
├── static    (静态文件)
│ ├── css
│ └── js
├── templates (网页文件)
│ └── index.html
├── init.py
├── app.py    (主文件)
├── public.py (公共方法)
└── config.py (配置文件)

结构说明

routers (路由控制)

控制路由的响应(静态文件，还是 Json, …)

响应静态文件 html.py

from fastapi import APIRouter, HTTPException
from starlette.templating import Jinja2Templates
from starlette.requests import Request

router = APIRouter()
# 绑定模板路径
template = Jinja2Templates(directory="templates")

@router.get("/")
async def index(request: Request):
    # request 为必传，后面的就跟 Jinja2 一样
    return template.TemplateResponse('index.html',{'request': request,"user":"啦啦啦啦"})

接口响应路由 users.py。/users/路径在本文件定义

from fastapi import APIRouter
router = APIRouter()

@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Foo"}, {"username": "Bar"}]

@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}

@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

接口响应路由 items.py。/items/路径在 app.py 定义

from fastapi import APIRouter, HTTPException

router = APIRouter()

@router.get("/")
async def use_api_munber():
    return [{"name": "Item Foo"}, {"name": "item Bar"}]

@router.get("/{item_id}")
async def read_item(item_id: str):
    return {"name": "Fake Specific Item", "item_id": item_id}

@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "foo":
        raise HTTPException(status_code=403, detail="You can only update the item: foo")
    return {"item_id": item_id, "name": "The Fighters"}

app.py (主文件)

from fastapi import Depends, FastAPI
from FastAPI_Project.routers import html, items, users
from public import count_use
from fastapi.staticfiles import StaticFiles
import config
import uvicorn

app = FastAPI(
    debug = True,
    version = config.VERSION,
)

# 静态文件绑定
app.mount('/static', StaticFiles(directory='static'), name='static')

# 静态文件路由
app.include_router(html.router)

# API 路由
app.include_router(
    items.router,
    prefix="/items", # 添加路由路径
    tags=["items"],  # 标签 路由分组作用
    # 注入依赖 请求这个路由之前,这个依赖的方法 可以用于验证 统计等
    dependencies=[Depends(count_use)],
)
app.include_router(
    users.router,
    dependencies=[Depends(count_use)],
    responses={404: {"description": "Not found"}},
)

if __name__ == '__main__':
    uvicorn.run(
        app= app,
        host= config.HOST,
        port= config.PORT,
    )

config.py 配置文件

配置参数，链接数据什么的

import pymongo
# 通用配置
class Config:
    # MongoDB 连接地址
    MONGODB = pymongo.MongoClient("mongodb://localhost:27017/")
    DATABASES = "bigdataboy"
    pass

# 运行地址
HOST = "127.0.0.1"
# 运行端口
PORT = 8686
# 版本
VERSION = "1.0.1"

public.py (公共方法)

from fastapi.logger import logger
from fastapi.requests import Request
from time import time
from FastAPI_Project.config import Config

"""
记录接口使用信息
"""
async def count_use(r: Request):
    # 获取api的路径
    path = r.url.path.strip("/").split("/")[0]
    # 获取客户端ip
    ip = r.client.host
    # 获取客户端 UA
    user_agent = r.headers.getlist("user-agent")[-1]
    logger.info(msg=f"{path} {ip} {user_agent}")
    # 插入 mongodb
    Config.MONGODB[Config.DATABASES]['count_use_api'].insert(
        {
            "path":path,                # 接口路径
             "ip":ip,                   # 客户端ip
             "user_agent":user_agent,   # 客户端UA
             "timestamp":time()         # 时间戳
        }
    )