FastAPI中的数据库应用介绍

---

FastAPI中的数据库应用介绍

FastAPI 是一个高性能和易用的日现应用框架，在构建数据库操作时同样具备高效性和完善性。本文将分为初级和高级应用，帮助你熟练从基础到高级的数据库操作。

---

一、初级应用

1.安装和配置

FastAPI 通常搭配 ORM 工具来操作数据库，最常用的 ORM 是 SQLAlchemy 和 Tortoise-ORM。下面以 SQLAlchemy 举例：

安装相关包

先通过 pip 安装必要包：

pip install fastapi[all] sqlalchemy mysql-connector-python

• mysql-connector-python: 用于操作 MySQL 数据库。

配置数据库

创建一个文件进行配置：

# database.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmakerDATABASE_URL = "mysql+mysqlconnector://username:password@localhost/dbname"engine = create_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

2. 创建数据表

创建数据表模型：

# models.py
from sqlalchemy import Column, Integer, String
from database import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)name = Column(String, index=True)email = Column(String, unique=True, index=True)

3. CRUD 操作

在 FastAPI 中设置 CRUD 操作功能：

创建文件

# crud.py
from sqlalchemy.orm import Session
from models import User# Create
def create_user(db: Session, name: str, email: str):db_user = User(name=name, email=email)db.add(db_user)db.commit()db.refresh(db_user)return db_user# Read
def get_user(db: Session, user_id: int):return db.query(User).filter(User.id == user_id).first()# Delete
def delete_user(db: Session, user_id: int):db_user = db.query(User).filter(User.id == user_id).first()if db_user:db.delete(db_user)db.commit()

绑定到 API 结构

为每个操作创建 API 结构：

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import SessionLocal, engine
from models import Base
from crud import create_user, get_userBase.metadata.create_all(bind=engine)app = FastAPI()def get_db():db = SessionLocal()try:yield dbfinally:db.close()@app.post("/users/")
def create_new_user(name: str, email: str, db: Session = Depends(get_db)):return create_user(db, name, email)@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):return get_user(db, user_id)

二、高级应用

1. 添加非阻塞操作

FastAPI 支持非阻塞操作，例如用 databases 完现非阻塞操作。

# async_database.py
from databases import DatabaseDATABASE_URL = "mysql+mysqlconnector://username:password@localhost/dbname"database = Database(DATABASE_URL)

在主要 API 中使用非阻塞操作：

@app.on_event("startup")
async def startup():await database.connect()@app.on_event("shutdown")
async def shutdown():await database.disconnect()@app.get("/users/async/{user_id}")
async def get_user_async(user_id: int):query = "SELECT * FROM users WHERE id = :user_id"return await database.fetch_one(query, values={"user_id": user_id})

2. 使用 Alembic 管理数据库模型迭代

Alembic 是一个用于管理数据库模型改动和迭代的工具。安装：

pip install alembic

创建 Alembic 配置：

alembic init alembic

配置 alembic.ini 文件，指定数据库连接信息：

sqlalchemy.url = mysql+mysqlconnector://username:password@localhost/dbname

创建数据表模型改动：

alembic revision --autogenerate -m "Add new user table"
alembic upgrade head

3. 实现处理更复杂的操作流

在实际项目中，层级分明是很重要的：

• 数据库的 DAO (Data Access Object)。
• 优化的非阻塞操作。
• 使用 Redis 和以缓存改善性能。

---