FastAPI 和数据库:使用 FastAPI 设置 SQLAlchemy

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示使用 Python 3.7+ 构建 API。构建 API 时最常见的任务之一是将其连接到数据库。在这篇博文中，我们将探讨如何使用 FastAPI 设置 SQLAlchemy 以与关系数据库交互。我们还将提供一些演示示例来帮助您入门。

为什么选择 SQLAlchemy？

SQLAlchemy 是一个功能强大的 SQL 工具包和 Python 对象关系映射 (ORM) 库。它提供了一整套众所周知的企业级持久性模式，旨在实现高效、高性能的数据库访问。

先决条件

在开始之前，请确保您已安装以下内容：

Python 3.7+

FastAPI

SQLAlchemy

数据库（例如，为简单起见，使用 SQLite）

您可以使用 pip 安装 FastAPI 和 SQLAlchemy：

pip install fastapi
pip install sqlalchemy
pip install uvicorn
pip install psycopg2-binary  # For PostgreSQL, replace with your database's driver

使用 FastAPI 设置 SQLAlchemy

1. 创建 FastAPI 项目结构

首先，让我们创建一个基本的项目结构：

fastapi_sqlalchemy_example/
    ├── app/
    │   ├── __init__.py
    │   ├── main.py
    │   ├── models.py
    │   ├── database.py
    │   ├── schemas.py
    │   └── crud.py
    └── requirements.txt

2. 定义数据库 URL

在app/database.py，我们将定义数据库连接：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # Replace with your database URL


engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


Base = declarative_base()

3. 创建模型

接下来，让我们在 app/models.py 中定义我们的数据库模型：

from sqlalchemy import Column, Integer, String
from .database import Base


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)
    hashed_password = Column(String)

4. 创建 CRUD 操作

在 app/crud.py 中，我们将定义与数据库交互的函数：

from sqlalchemy.orm import Session
from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(name=user.name, email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

5. 定义模式

在 app/schemas.py 中，为数据定义 Pydantic 模型：

from pydantic import BaseModel


class UserBase(BaseModel):
    name: str
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int


    class Config:
        orm_mode = True

6. 创建依赖项

在 app/main.py 中，我们创建数据库会话依赖项：

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine


models.Base.metadata.create_all(bind=engine)


app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

7. 创建 API 端点

最后，让我们在 app/main.py 中创建 API 端点：

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

8. 运行应用程序

要运行 FastAPI 应用程序，请使用 Uvicorn：

uvicorn app.main:app --reload

以下是一些基于上述基本设置的更高级的演示，展示了附加功能和最佳实践。

演示 1：添加模型之间的关系

让我们添加一个名为 Item 的新模型，该模型与 User 模型有关系。

步骤 1：更新 models.py

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base


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)
    hashed_password = Column(String)


    items = relationship("Item", back_populates="owner")




class Item(Base):
    __tablename__ = "items"


    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))


    owner = relationship("User", back_populates="items")

步骤 2：更新 schemas.py

from typing import List, Optional
from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int


    class Config:
        orm_mode = True




class UserBase(BaseModel):
    name: str
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    items: List[Item] = []


    class Config:
        orm_mode = True

步骤 3：更新 crud.py

from sqlalchemy.orm import Session
from . import models, schemas


# Existing CRUD functions...


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


def get_items(db: Session, skip: int = 0, limit: int = 10):
    return db.query(models.Item).offset(skip).limit(limit).all()

步骤 4：使用新端点更新 main.py

from typing import List


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

演示 2：使用 OAuth2 添加身份验证

步骤 1：安装依赖项

pip install python-jose[cryptography]
pip install passlib[bcrypt]

步骤 2：为安全实用程序创建 auth.py

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session


from . import crud, models, schemas
from .database import SessionLocal


# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def authenticate_user(db: Session, email: str, password: str):
    user = crud.get_user_by_email(db, email=email)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = crud.get_user_by_email(db, email=email)
    if user is None:
        raise credentials_exception
    return user

步骤 3：更新 crud.py 以使用散列密码

from .auth import get_password_hash


def create_user(db: Session, user: schemas.UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = models.User(name=user.name, email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

步骤 4：在 main.py 中添加令牌端点和安全路由

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from datetime import timedelta
from .auth import authenticate_user, create_access_token, get_current_user


@app.post("/token", response_model=schemas.Token)
async def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.email}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=schemas.User)
async def read_users_me(current_user: schemas.User = Depends(get_current_user)):
    return current_user

步骤 5：更新 schemas.py 以支持 OAuth2

from pydantic import BaseModel


class Token(BaseModel):
    access_token: str
    token_type: str

演示 3：分页和过滤

步骤 1：在 crud.py 中为 CRUD 操作添加分页

def get_users(db: Session, skip: int = 0, limit: int = 10, search: str = None):
    query = db.query(models.User)
    if search:
        query = query.filter(models.User.name.contains(search))
    return query.offset(skip).limit(limit).all()

步骤 2：在 main.py 中为端点添加分页和过滤

@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 10, search: str = None, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit, search=search)
    return users

这些额外的演示展示了更高级的用例，例如在模型之间添加关系、使用 OAuth2 实现身份验证以及向端点添加分页和过滤。这些功能中的每一个都增强了 FastAPI 应用程序的功能和可用性。您可以随意进一步定制和扩展这些示例以满足您的特定需求。

在这篇博文中，我们用 SQLAlchemy 设置了一个 FastAPI 应用程序来与关系数据库交互。我们介绍了如何定义模型、创建 CRUD 操作以及设置 API 端点。这只是一个起点，您也可以模仿上述添加更复杂的查询、处理模型之间的关系以及实现更高级的功能来扩展此设置。