当前位置：首页 > news >正文

Python使用总结之FastAPI高级功能探索：数据库集成与依赖注入

news 2024/11/17 8:21:48

Python使用总结之FastAPI高级功能探索：数据库集成与依赖注入

FastAPI不仅在构建简单的API服务时表现出色，还提供了多种高级功能，帮助开发者应对更复杂的应用场景。本文将深入探讨FastAPI的高级特性，特别是数据库集成与依赖注入。

一、数据库集成

对于大多数Web应用来说，数据库是不可或缺的部分。FastAPI可以轻松与ORM（对象关系映射）工具如SQLAlchemy、Tortoise ORM等集成，来处理关系型数据库。以下将以SQLAlchemy为例，展示如何在FastAPI中实现数据库集成。

1. 安装依赖

首先安装SQLAlchemy、databases（用于异步支持）和aiosqlite等依赖：

pip install sqlalchemy databases aiosqlite

2. 定义数据库模型

使用SQLAlchemy定义数据库模型：

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerDATABASE_URL = "sqlite:///./test.db"Base = declarative_base()
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)class 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. 创建数据库连接

使用databases库进行异步数据库操作：

from databases import Databasedatabase = Database(DATABASE_URL)

4. 在FastAPI中使用数据库

通过依赖注入的方式将数据库连接注入到FastAPI的路由中。下面展示了如何在API端点中操作数据库：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Sessionapp = FastAPI()# Dependency to get a SQLAlchemy session
def get_db():db = SessionLocal()try:yield dbfinally:db.close()@app.on_event("startup")
async def startup():await database.connect()@app.on_event("shutdown")
async def shutdown():await database.disconnect()@app.get("/users/{user_id}")
async def read_user(user_id: int, db: Session = Depends(get_db)):user = db.query(User).filter(User.id == user_id).first()return user

在这个例子中，我们通过get_db()函数将数据库会话注入到API的路由中，使用Depends确保每个请求都会创建和关闭会话，保证了数据库操作的安全和一致性。

二、依赖注入

FastAPI的依赖注入系统极大地提高了代码的模块化和可复用性。通过依赖注入，开发者可以轻松管理复杂应用中的共享资源或状态，如数据库连接、配置参数、用户认证等。

1. 简单依赖注入示例

依赖注入可以用于向API端点传递任何需要的共享对象或数据。以下示例展示了如何通过依赖注入共享配置参数：

from fastapi import Dependsclass Settings:app_name: str = "Advanced FastAPI App"admin_email: str = "admin@example.com"def get_settings():return Settings()@app.get("/info")
def get_info(settings: Settings = Depends(get_settings)):return {"app_name": settings.app_name, "admin_email": settings.admin_email}

在这里，get_settings()函数被注册为依赖项，Depends将其结果注入到get_info路由中。

2. 带有状态的依赖

依赖项可以包含复杂的逻辑或状态。比如，如果你需要为每个请求生成不同的访问令牌，依赖注入可以帮助管理这个逻辑：

from fastapi import Requestclass TokenGenerator:def __init__(self, secret_key: str):self.secret_key = secret_keydef generate_token(self, request: Request):return f"token-for-{request.client.host}"def get_token_generator():return TokenGenerator(secret_key="mysecret")@app.get("/token")
def get_token(request: Request, token_generator: TokenGenerator = Depends(get_token_generator)):token = token_generator.generate_token(request)return {"token": token}

通过这种方式，你可以在复杂的应用中共享并管理依赖，确保逻辑的集中管理和灵活性。

三、数据库事务与依赖注入结合

在实际应用中，事务的管理是确保数据库一致性的重要部分。我们可以结合数据库集成和依赖注入的概念，确保在每个请求中正确管理数据库事务。

from fastapi import HTTPException, statusdef get_db():db = SessionLocal()try:yield dbdb.commit()  # 提交事务except:db.rollback()  # 发生错误时回滚事务raisefinally:db.close()@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):db_user = User(email=user.email, name=user.name)db.add(db_user)db.commit()  # 手动提交事务db.refresh(db_user)return db_user