FastAPI 是一个现代、高性能的 Web 框架，基于 Python 3.7+，用于构建 API，它使用标准的 Python 类型提示。SQLAlchemy 是一个强大的 SQL 工具包和 Python 的对象关系映射 (ORM) 库。本文将探讨如何使用 SQLAlchemy 将关系型数据库与 FastAPI 集成在一起，并通过实际代码示例一步步指导您完成整个集成过程。
1. 搭建您的开发环境首先，你需要安装必要的依赖；接着，用 pip 即可完成安装。
pip install fastapi uvicorn sqlalchemy asyncpg psycopg2-binary这里有一个你FastAPI项目的简单框架。
项目结构如下:
my_project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── models.py
│ ├── crud.py
│ ├── schemas.py
│ └── database.py
├── requirements.txt
└── README.md创建一个名为 database.py 的文件来建立数据库连接和会话对象。
例子:**database.py**
从 sqlalchemy 导入 create_engine # 导入 create_engine 函数
从 sqlalchemy.ext.declarative 导入 declarative_base # 导入 declarative_base 类
从 sqlalchemy.orm 导入 sessionmaker # 导入 sessionmaker 函数
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname" # 数据库连接字符串
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 创建会话工厂
Base = declarative_base() # 声明基类
def get_db(): # 获取数据库会话
db = SessionLocal() # 创建会话
try:
yield db
finally: # 最后关闭会话
db.close()在 models.py 文件里定义你的数据库模型。这些模型对应数据库中的表。
例子:**models.py**
从 sqlalchemy 导入 Column, Integer, String
从 .database 导入 基础类 Base # ORM基类
# 定义一个名为Item的数据库模型
class Item(Base):
# 设置表名为items
表名 = "items"
# 定义一个整数类型的列，作为主键并建立索引，用于唯一标识每一项
id = Column(Integer, primary_key=True, index=True)
# 定义一个字符串类型的列，并建立索引
name = Column(String, index=True)
# 定义一个字符串类型的列，并建立索引
description = Column(String, index=True)使用 Pydantic 模式定义来验证并序列化数据。你可以在这一步中，在 schemas.py 中创建这些数据模式。
例子：**schemas.py**
from pydantic import BaseModel
# ItemBase类定义了一个基础模型，包含名称和描述。
class ItemBase(BaseModel):
name: str
description: str
# ItemCreate类继承自ItemBase，定义了创建项的基础模型。
class ItemCreate(ItemBase):
pass
# Item类继承自ItemBase，定义了项的详细模型，包含id字段。
class Item(ItemBase):
id: int
# Config类定义了orm模式为True，表示使用ORM模式。
class Config:
orm_mode = True在 crud.py 文件中定义 CRUD 操作 (创建、读取、更新、删除)。
例子：crud.py
from sqlalchemy.orm import Session
from . import models, schemas
def get_item(db: Session, item_id: int):
"""
获取指定ID的项目
"""
return db.query(models.Item).filter(models.Item.id == item_id).first()
def get_items(db: Session, skip: int = 0, limit: int = 10):
"""
获取偏移量为skip，限制为limit的项目列表
"""
return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate):
"""
创建一个新的项目
"""
db_item = models.Item(name=item.name, description=item.description)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item- API接口的创建.
在 main.py 文件中定义你的 FastAPI API 端点以便与数据库进行交互。
例子:main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine, get_db
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.post("/items/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
return crud.create_item(db=db, item=item)
@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
db_item = crud.get_item(db, item_id=item_id)
if db_item is None:
raise HTTPException(status_code=404, detail="找不到该项目")
return db_item
@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你可以使用 Uvicorn 启动你的 FastAPI 应用程序。
uvicorn app.main:app --reload这是一个命令，用于使用Uvicorn ASGI服务器运行Python应用程序。
导航到 http://127.0.0.1:8000 即可访问您的API。FastAPI 还提供了一个交互式的API文档页面，可通过 [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs) 访问。
将FastAPI与SQLAlchemy集成可以让你创建功能强大且高性能的API，并使用关系型数据库作为后端。按照上述步骤，你可以搭建一个稳健的FastAPI应用，利用SQLAlchemy实现高效的数据库操作。
了解最新的 FastAPI 技巧和最佳实践，紧跟技术潮流！关注我们的通讯，获取更多见解、技巧和独家内容！
现在就订阅，让FastAPI项目更上一层楼！