FastAPI整合SQLModel实现增删改查接口

安装依赖

pip install fastapi "uvicorn[standard]" sqlmodel

定义模型

from sqlmodel import Field, Session, SQLModel, create_engine, select

class User(SQLModel, table=True):

"""用户表"""

id: int | None = Field(default=None, primary_key=True)

name: str = Field(index=True)

age: int = Field(default=None, index=True)

创建表

from fastapi import FastAPI

from sqlmodel import Field, Session, SQLModel, create_engine, select

class User(SQLModel, table=True):

"""用户表"""

id: int | None = Field(default=None, primary_key=True)

name: str = Field(index=True)

age: int = Field(default=None, index=True)

app = FastAPI()

db = "db.sqlite3"

db_url = "sqlite:///" + db

connect_args = {"check_same_thread": False}

engine = create_engine(db_url, echo=True, connect_args=connect_args)

@app.on_event("startup")

def on_startup():

# 创建表

SQLModel.metadata.create_all(engine)

新增用户

@app.post("/user")

async def add_user(user: User):

"""新增用户的接口"""

with Session(engine) as session:

session.add(user)

session.commit()

session.refresh(user)

return user

获取所有用户

@app.get("/user")

async def get_user_all():

"""获取所有用户的接口"""

with Session(engine) as session:

users = session.exec(select(User)).all()

return users

完整代码

from fastapi import FastAPI

from sqlmodel import Field, Session, SQLModel, create_engine, select

class User(SQLModel, table=True):

"""用户表"""

id: int | None = Field(default=None, primary_key=True)

name: str = Field(index=True)

age: int = Field(default=None, index=True)

app = FastAPI()

db = "db.sqlite3"

db_url = "sqlite:///" + db

connect_args = {"check_same_thread": False}

engine = create_engine(db_url, echo=True, connect_args=connect_args)

@app.on_event("startup")

def on_startup():

# 创建表

SQLModel.metadata.create_all(engine)

@app.post("/user")

async def add_user(user: User):

"""新增用户的接口"""

with Session(engine) as session:

session.add(user)

session.commit()

session.refresh(user)

return user

@app.get("/user")

async def get_user_all():

"""获取所有用户的接口"""

with Session(engine) as session:

users = session.exec(select(User)).all()

return users

if __name__ == "__main__":

import uvicorn

uvicorn.run(app)