下面是一个使用 FastAPI 和 SQLAlchemy 连接 SQLite 数据库的简单示例：
首先，确保安装以下依赖：
pip install fastapi uvicorn sqlalchemy databases[sqlite] alembic
然后，创建一个 main.py 文件：
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from databases import Database
DATABASE_URL = "sqlite:///./test.db"
DATABASE_ENGINE = create_engine(DATABASE_URL)
# 使用 FastAPI 创建一个应用
app = FastAPI()
# 使用 databases 来处理异步数据库连接
database = Database(DATABASE_URL)
# 使用 SQLAlchemy 的 declarative_base 创建基本模型
Base = declarative_base()
# 定义模型类，表示数据库中的表
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String)
# 创建数据库表
Base.metadata.create_all(bind=DATABASE_ENGINE)
# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=DATABASE_ENGINE)
# 依赖项，用于获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 路径操作函数，用于创建新项目
@app.post("/items/")
async def create_item(item: Item, db: Session = Depends(get_db)):
db.add(item)
db.commit()
db.refresh(item)
return item
# 路径操作函数，用于获取项目列表
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
items = db.query(Item).offset(skip).limit(limit).all()
return items
在这个例子中：
- 使用 SQLAlchemy 定义了一个简单的数据模型 Item，表示数据库中的 items 表。
- 使用 databases.Database 来处理异步数据库连接，以及 sqlalchemy.create_engine 创建数据库引擎。
- 使用 SessionLocal 来创建数据库会话。
- 使用 Depends 作为依赖项，通过 get_db 函数获取数据库会话。
- 定义了两个路径操作函数，一个用于创建新项目，一个用于获取项目列表。
这只是一个简单的示例，实际中你可能需要更复杂的模型和操作。FastAPI 配合 SQLAlchemy 提供了强大的数据库支持，你可以使用 ORM 来轻松操作数据库。
转载请注明出处：http://www.pingtaimeng.com/article/detail/7396/FastAPI