引言
FastAPI 是一个现代、快速(高性能)的 Web 框架,用于构建 API,由 Starlette 和 Pydantic 驱动。它支持异步请求处理,使得开发高性能的 Web 应用成为可能。在本文中,我们将深入探讨如何使用 FastAPI 与数据库进行交互,并提供一些实战案例,帮助您轻松掌握高效开发技巧。
快速了解FastAPI
在开始之前,让我们先快速了解 FastAPI 的一些关键特性:
- 异步支持:FastAPI 使用异步函数来处理请求,这可以显著提高性能。
- 自动文档:FastAPI 自动生成交互式 API 文档,方便开发者测试和调试。
- 类型安全:FastAPI 利用 Pydantic 进行数据验证,确保输入数据的正确性。
- 依赖注入:FastAPI 支持依赖注入,简化了代码的编写和维护。
与数据库交互
FastAPI 支持多种数据库,如 PostgreSQL、MySQL、SQLite 等。以下是一些常用的数据库交互方法:
1. 使用SQLAlchemy
SQLAlchemy 是一个强大的 SQL 工具包和对象关系映射(ORM)系统。以下是使用 SQLAlchemy 与 PostgreSQL 交互的示例:
from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 定义数据库模型
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
age = Column(Integer)
# 创建数据库引擎
engine = create_engine('postgresql://user:password@localhost/dbname')
# 创建会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
app = FastAPI()
# 获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
return db_user
raise HTTPException(status_code=404, detail="User not found")
2. 使用Gino
Gino 是一个轻量级的异步 ORM,专门为 PostgreSQL 设计。以下是使用 Gino 与数据库交互的示例:
from fastapi import FastAPI
from gino import Gino
# 定义数据库模型
db = Gino()
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String)
age = db.Column(db.Integer)
app = FastAPI()
@app.get("/users/{user_id}")
def read_user(user_id: int):
db_user = User.get(user_id)
if db_user:
return db_user
raise HTTPException(status_code=404, detail="User not found")
实战案例:用户管理系统
以下是一个简单的用户管理系统示例,展示了如何使用 FastAPI 和 SQLAlchemy 创建、读取、更新和删除用户:
from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 定义数据库模型
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
age = Column(Integer)
# 创建数据库引擎
engine = create_engine('postgresql://user:password@localhost/dbname')
# 创建会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
app = FastAPI()
# 获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/users/")
def create_user(name: str, age: int, db: Session = Depends(get_db)):
db_user = User(name=name, age=age)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
return db_user
raise HTTPException(status_code=404, detail="User not found")
@app.put("/users/{user_id}")
def update_user(user_id: int, name: str, age: int, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
db_user.name = name
db_user.age = age
db.commit()
db.refresh(db_user)
return db_user
raise HTTPException(status_code=404, detail="User not found")
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
db.delete(db_user)
db.commit()
return {"detail": "User deleted successfully"}
raise HTTPException(status_code=404, detail="User not found")
总结
通过本文,您已经了解了如何使用 FastAPI 与数据库进行交互,并掌握了一些高效开发技巧。在实际项目中,您可以根据需求选择合适的数据库和 ORM 框架,并结合 FastAPI 的强大功能,轻松构建高性能的 Web 应用。祝您在 FastAPI 之旅中一切顺利!
