fastapi 新增数据库接口

### 如何在 FastAPI 中新增数据库接口的最佳实践 #### 1. 安装依赖库 为了使 FastAPI 应用程序能够连接并操作数据库，通常需要安装相应的 ORM 或者驱动工具。SQLAlchemy 是一种常用的 Python SQL 工具包和对象关系映射器 (ORM)，可以很好地与 FastAPI 集成。 ```bash pip install fastapi sqlalchemy databases uvicorn ``` #### 2. 创建项目结构 合理的文件夹布局有助于维护项目的清晰度： ``` project/ │ ├── app/ │ ├── __init__.py │ ├── main.py # 主应用入口 │ ├── models.py # 数据库模型定义 │ ├── schemas.py # Pydantic 模型验证 │ └── crud.py # CRUD 操作逻辑封装 └── alembic/ # 数据库迁移配置（可选） └── ... ``` #### 3. 初始化 SQLAlchemy 和 Databases 实例 `main.py` 文件中初始化必要的组件来管理数据库会话以及异步查询支持。 ```python from typing import AsyncGenerator, Generator import databases import sqlalchemy from fastapi import Depends, FastAPI from pydantic import BaseModel DATABASE_URL = "sqlite:///./test.db" database = databases.Database(DATABASE_URL) metadata = sqlalchemy.MetaData() engine = sqlalchemy.create_engine( DATABASE_URL, connect_args={"check_same_thread": False} ) def get_db() -> Generator: try: yield database finally: pass app = FastAPI() @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() ``` #### 4. 定义数据表模式 通过 `models.py` 来描述实体类及其字段属性，并将其映射到实际的关系型表格上。 ```python from sqlalchemy import Column, Integer, String, Table, ForeignKey from .main import metadata users_table = Table( 'users', metadata, Column('id', Integer, primary_key=True), Column('name', String(50)), Column('email', String(100)) ) ``` #### 5. 构建 API 请求体校验规则 `schemas.py` 使用 Pydantic 提供的数据解析功能确保传入参数的有效性和一致性。 ```python from pydantic import BaseModel, EmailStr class UserCreate(BaseModel): name: str email: EmailStr class UserRead(UserCreate): id: int class Config: orm_mode = True ``` #### 6. 封装基本增删改查函数 `crud.py` 负责处理具体的业务流程，比如插入新纪录、检索现有条目等。 ```python from sqlalchemy.sql import select from sqlalchemy.engine.base import Connection from .schemas import UserCreate from .models import users_table async def create_user(conn: Connection, user: UserCreate) -> dict: query = users_table.insert().values(**user.dict()) result = await conn.execute(query) return {**user.dict(), "id": result.lastrowid} async def read_users(conn: Connection) -> list[dict]: query = select([users_table]) results = await conn.fetch_all(query=query) return [{column: value for column, value in row.items()} for row in results] ``` #### 7. 设计 RESTful 接口路由 最后，在 `main.py` 添加路径处理器完成对外服务暴露。 ```python from fastapi import HTTPException from starlette.status import HTTP_404_NOT_FOUND from .crud import create_user, read_users from .schemas import UserCreate, UserRead @app.post("/users/", response_model=UserRead) async def post_create_user(user: UserCreate, db=Depends(get_db)): new_user = await create_user(db, user=user) return new_user @app.get("/users/") async def get_read_users(db=Depends(get_db)): all_users = await read_users(db) if not all_users: raise HTTPException(status_code=HTTP_404_NOT_FOUND, detail="No Users Found.") return {"data": all_users} ``` 上述过程展示了如何构建一个简单的 FastAPI Web Service 并集成 SQLite 数据库存储能力[^1]。当然，具体实现细节可能因所选用的不同类型的数据库而有所差异；此外，考虑到生产环境下的性能优化需求，还可以引入诸如缓存机制、分页加载等功能特性进一步增强系统的健壮性。