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

代码

import time
from typing import Annotatedfrom fastapi import FastAPI, Depends, Query, status
from fastapi.exceptions import HTTPException
from pydantic import BaseModel
from sqlmodel import Field, Session, SQLModel, create_engine, selectapp = 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 = Noneage: int | None = Nonesecret_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