fastapi通过sqlmodel连接Mysql实现crud功能

代码

import time
from typing import Annotated

from fastapi import FastAPI, Depends, Query, status
from fastapi.exceptions import HTTPException
from pydantic import BaseModel
from sqlmodel import Field, Session, SQLModel, create_engine, select

app = FastAPI()

# 数据库连接URL
DATABASE_URL = f"mysql+pymysql://root:password@localhost:3306/demo"
# 创建数据库引擎
engine = create_engine(DATABASE_URL, echo=True)  # echo=True 用于调试，显示SQL语句


# 英雄基类
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)


# 定义数据模型
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str


# 响应模型
class HeroPublic(HeroBase):
    id: int


# 创建模型
class HeroCreate(HeroBase):
    secret_name: str


# 更新模型
class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None


# 创建数据库表
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


# 获取数据库会话
def get_session():
    with Session(engine) as session:
        yield session


# 启动时创建数据库表
@app.on_event("startup")
def on_startup():
    create_db_and_tables()


SessionDep = Annotated[Session, Depends(get_session)]


# 添加英雄路由
@app.post("/heroes/", response_model=HeroPublic)
async def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


# 获取所有英雄路由
@app.get("/heroes/", response_model=list[HeroPublic])
async def read_heroes(*, skip: int = 0, limit: Annotated[int, Query(le=100)] = 100, session: SessionDep):
    heroes = session.exec(select(Hero).offset(skip).limit(limit)).all()
    return heroes


# 获取单个英雄路由
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
async def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero


# 更新单个英雄
@app.put("/heroes/{hero_id}", response_model=HeroPublic)
async def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db


# 删除单个英雄
@app.delete("/heroes/{hero_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return

依赖

annotated-types==0.7.0
anyio==4.10.0
bcrypt==4.3.0
certifi==2025.8.3
click==8.2.1
colorama==0.4.6
dnspython==2.7.0
email-validator==2.3.0
fastapi==0.116.1
fastapi-cli==0.0.8
fastapi-cloud-cli==0.1.5
greenlet==3.2.4
h11==0.16.0
httpcore==1.0.9
httptools==0.6.4
httpx==0.28.1
idna==3.10
Jinja2==3.1.6
markdown-it-py==4.0.0
MarkupSafe==3.0.2
mdurl==0.1.2
mysql-connector-python==9.4.0
passlib==1.7.4
pydantic==2.11.7
pydantic_core==2.33.2
Pygments==2.19.2
PyJWT==2.10.1
PyMySQL==1.1.2
python-dotenv==1.1.1
python-multipart==0.0.20
PyYAML==6.0.2
rich==14.1.0
rich-toolkit==0.15.0
rignore==0.6.4
sentry-sdk==2.35.1
shellingham==1.5.4
sniffio==1.3.1
SQLAlchemy==2.0.43
sqlmodel==0.0.24
starlette==0.47.3
tqdm==4.67.1
typer==0.17.3
typing-inspection==0.4.1
typing_extensions==4.15.0
urllib3==2.5.0
uvicorn==0.35.0
watchfiles==1.1.0
websockets==15.0.1