Python fastapi

介绍

FastAPI 是一个现代、快速、开源的 Web 框架，用于构建高性能的 RESTful API。

安装

 $ python -m pip install "fastapi[all]"

 tree .
├── config
├── lib                                          # 存放自定义模块(无变动或变动少)
│   └── common.py 
├── LICENSE
├── log                                          # 存放日志文件
├── temp                                         # 存放临时文件
├── static                                       # 静态文件
├── .gitignore                                   # git 忽略文件
├── main.py                                      # 启动文件
├── README.md 
├── requirement.txt                              # 项目依赖包
├── api                                          # api 路由
│   └── v1
│       └── node.py 
└── utils                                        # 自定义模块
    ├── database.py                              # 数据库连接
    ├── logger.py                                # logger 初始化
    ├── middleware.py                            # 中间件
    ├── router.py                                # 路由配置
    └── schemas.py                               # fastapi 参数模型

fastapi 入口

from contextlib import asynccontextmanager

import uvicorn
from loguru import logger
from fastapi import FastAPI

from utils.logger import logger_setting
from utils.router import add_routers
from utils.database import Database
from utils.middleware import add_middlewares

HOST = "localhost"
PORT = 8010

@asynccontextmanager
async def lifespan(router: FastAPI):
    """ 应用开启和结束操作 """
    logger_setting()
    Database.init()
    add_routers(router)
    logger.info("app start")
    yield
    logger.info("app close")
    Database.close()

app = FastAPI(
    title="FastAPI",
    description=f"{HOST}:{PORT} api",
    version="0.0.1",
    lifespan=lifespan
)

add_middlewares(app)


if __name__ == "__main__":
    uvicorn.run("main:app", host=HOST, port=PORT, reload=True)

浏览器打开 http://0.0.0.0:8010/
swagger-ui http://0.0.0.0:8010/docs/

swagger-ui

Github swagger-ui

打开 swagger 页面需要加载外界 CDN 资源，可能会非常慢，建议使用离线 swagger-ui 文件
在 Github 下载官方 swagger-ui 资源, 放入项目 static 路径下, 在代码中配置即可

project
├─ static
│  └── swagger-ui
│      ├── ...
│      └── README.md
└── main.py

在 utils/router.py 中挂载 swagger 并配置可用路由

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.openapi.docs import get_swagger_ui_html

from api.v1 import student, article
from lib.common import abs_dir

def add_routers(app: FastAPI):
    """ app add router"""
    # 添加路由
    app.include_router(student.router, prefix="/api/v1", tags=["student"])
    app.include_router(article.router, prefix="/api/v1", tags=["article"])
    
    # mount swagger 静态文件
    app.mount('/static', StaticFiles(
        directory=abs_dir("static", "swagger-ui", "dist")),
        name="static"
    )

    # 定义 swagger 挂载路由
    @app.get("/", include_in_schema=False)
    async def custom_swagger_ui_html():
        """ set local static swagger """
        return get_swagger_ui_html(openapi_url=app.openapi_url, title=app.title)

database

定义数据连接和使用

from contextlib import contextmanager
from dataclasses import dataclass

import psycopg
from psycopg_pool import ConnectionPool
from psycopg.rows import dict_row
from loguru import logger


@dataclass
class DBConnect:
    """ 数据库连接 """
    host: str
    port: int
    dbname: str
    user: str
    password: str

DEFAULT_DATABASE: DBConnect = DBConnect(
    host="localhost",
    port="5432",
    dbname="learn",
    user="postgres",
    password="password"
)

class Database:
    """ database module """
    pool: ConnectionPool| None = None
    db: DBConnect = DEFAULT_DATABASE

    @classmethod
    def init(cls, db: DBConnect|None=None) -> None:
        """ 数据库初始化连接 """
        cls.db = db if db else cls.db
        cls.pool: ConnectionPool = ConnectionPool(
            min_size=1,
            max_size=10,
            conninfo=" ".join(f"{k}={v}" for k, v in vars(cls.db).items())
        )

    @contextmanager
    def __new__(cls, db: DBConnect|None=None):
        _ = cls.init(db) if db else None
        try:
            conn = cls.pool.getconn()
            cursor = conn.cursor(row_factory=dict_row)
            yield cursor
            conn.commit()
        except Exception as e:
            _ = conn.rollback() if conn else None
            logger.error(f"error: {e}")
        finally:
            _ = cursor.close() if cursor else None
            _ = cls.pool.putconn(conn) if cls.pool else None


if __name__ == "__main__":
    # 起服务时连接数据库
    Database.init()
    # 使用连接池执行 sql 语句
    with Database() as cursor:
        cursor.execute("SELECT id, title FROM article")
        print(cursor.fetchall())

api 路由

定义 fastapi api 路由

from fastapi import APIRouter, Depends
from loguru import logger
from psycopg import Cursor

from utils.database import Database
from utils.schemas import Student

router = APIRouter()
created = Database.create_session

@router.get("/articles")
def get_articles(session: Cursor = Depends(created)):
    return session.execute("SELECT * FROM article").fetchall()

@router.get("/articles/{id}")
def get_article(id: int, session: Cursor = Depends(created)):
    return session.execute("SELECT * FROM article WHERE id = %s", (id,)).fetchone()

日志

设置日志打印格式, 输出到指定文件

from loguru import logger

from lib.common import abs_dir

def logger_setting():
    """ 设置 logger """
    logger.remove()

    fmt = '[<level>{level: <8}</level>][<green>{time:YYYY-MM-DD HH:mm:ss}</green>]: <level>{message}</level>'
    logger.add(sys.stderr,  level='INFO', format=fmt)
    logger.add(abs_dir('log', 'report.log'),
        level='INFO', format=fmt, rotation='1 week', retention='30 days'
    )

中间件

自定义请求开始和请求结束后的操作

from traceback import format_exc

from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from loguru import logger


def add_middlewares(app: FastAPI):
    """ 允许所有请求 """
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    @app.middleware("http")
    async def catch_exceptions(request: Request, call_next):
        """ 捕获所有接口执行异常 """
        try:
            response = await call_next(request)
            return response
        except Exception as e:
            logger.error(format_exc())
            return JSONResponse(status_code=500, content={"message": "Server run error"})

Python fastapi

Table of Contents

介绍

安装

fastapi 入口

swagger-ui

database

api 路由

日志

中间件