SQLModel

起步

安裝

pip install sqlmodel

# mysql驅(qū)動(dòng)
pip install pymysql
# postgresql驅(qū)動(dòng)
pip install psycopg2-binary

連接引擎

from sqlmodel import SQLModel, create_engine

# 內(nèi)存
db_url = "sqlite+pysqlite:///:memory:"
# sqlite
db_url = "sqlite+pysqlite:///db_demo.sqlite"
# mysql
db_url = "mysql+pymysql://username:password@127.0.0.1:3306/db_name"
# postgresql
db_url = "postgresql://username:password@127.0.0.1:5432/db_name"

# 創(chuàng)建引擎
engine = create_engine(db_url, echo=True)

# 創(chuàng)建表格
SQLModel.metadata.create_all(engine)

數(shù)據(jù)表

import time
from sqlmodel import SQLModel, create_engine
from sqlmodel import Field, BigInteger, Integer, String, Boolean

# 直接創(chuàng)建
class User(SQLModel, table=True):
    __tablename__ = "user"
    id: int | None = Field(sa_type=BigInteger, default=None, primary_key=True)
    name: str = Field(sa_type=String(32), description="用戶名", sa_column_kwargs={"comment": "用戶名"})


# 使用基類
class BaseTable(SQLModel):
    id: int | None = Field(sa_type=BigInteger, default=None, primary_key=True)
    create_timestamp: int = Field(
        sa_type=BigInteger,
        default_factory=lambda: time.time_ns() // 1000000,  # lambda使用default_factory
        description="創(chuàng)建時(shí)間戳",
        sa_column_kwargs={"comment": "創(chuàng)建時(shí)間戳"},
    )

class UseBaseUser(BaseTable, table=True):
    __tablename__ = "use_base_user"
    name: str = Field(sa_type=String(32), sa_column_kwargs={"comment": "用戶名"})
    age: int = Field(sa_type=Integer, default=0, ge=0, le=200, sa_column_kwargs={"comment": "年齡"})
字段 作用 示例 備注
tablename 自定義表名 "user" 可省略
sa_type 數(shù)據(jù)表類型 BigInteger sa即SqlAlchemy
sa_column_kwargs 數(shù)據(jù)表參數(shù) {"comment": "用戶名"} sa即SqlAlchemy，可填寫(xiě)數(shù)據(jù)庫(kù)的字段備注
文檔：https://github.com/fastapi/sqlmodel/issues/492
default 默認(rèn)值 None
default_factory 默認(rèn)值工廠 lambda: ...
primary_key 主鍵 True
title 標(biāo)題 用于Swagger等
description 標(biāo)題 用于Swagger等
gt/ge/lt/le 范圍 pydantic的校驗(yàn)

JSON數(shù)據(jù)

  • sa_type為JSON，保存和獲取都能使用對(duì)應(yīng)的類型
from sqlmodel import SQLModel
from sqlmodel import Field, BigInteger, Integer, String, Boolean, JSON

# 直接創(chuàng)建
class User(SQLModel, table=True):
    __tablename__ = "user"
    id: int | None = Field(sa_type=BigInteger, default=None, primary_key=True)
    name: str = Field(sa_type=String(32), description="用戶名", sa_column_kwargs={"comment": "用戶名"})
    
    # JSON格式，數(shù)據(jù)表中為json格式，存取都是對(duì)應(yīng)的類型(list、dict)
    addrs: list[str] = Field(sa_type=JSON, default=[], sa_column_kwargs={"comment": "地址列表"})
    other: dict = Field(sa_type=JSON, default={}, sa_column_kwargs={"comment": "其他信息"})

# 插入數(shù)據(jù)庫(kù)不需要特殊處理
user = User(name="李四", addrs=["福州"], other={"a": 1, "b": 2})
with Session(engine) as session:
    session.add(user)
    session.commit()
    session.refresh(entity)

# 讀取結(jié)果也能自動(dòng)轉(zhuǎn)換
with Session(engine) as session:
    statement = select(User)
    result = session.exec(statement).all()
    for i in result:
        print(type(i.addrs), i.addrs)
        # <class 'list'> ['福州']
        print(type(i.other), i.other)
        # <class 'dict'> {'a': 1, 'b': 2}

枚舉類

  • 當(dāng)前版本枚舉類較為雞肋，無(wú)法自動(dòng)轉(zhuǎn)換，需要轉(zhuǎn)換成字符串，否則報(bào)錯(cuò)
  • 而且不會(huì)校驗(yàn)字符串是否是枚舉類
from enum import Enum
from sqlmodel import SQLModel
from sqlmodel import Field, BigInteger, String

# 枚舉類型
class SexEnum(Enum):
    MALE = "MALE"
    FEMALE = "FEMALE"
    UNKNOWN = "UNKNOWN"


class User(SQLModel, table=True):
    __tablename__ = "user"
    id: int | None = Field(sa_type=BigInteger, default=None, primary_key=True)
    name: str = Field(sa_type=String(32), default="")
    
    # 枚舉字段;數(shù)據(jù)庫(kù)類型為字符串;不會(huì)自動(dòng)轉(zhuǎn)換，傳入?yún)?shù)要加上.value轉(zhuǎn)成字符串(default也一樣)
    sex: SexEnum = Field(sa_type=String(32), default=SexEnum.UNKNOWN.value)

# sex參數(shù)需要轉(zhuǎn)換成字符串
user = User(name="zhangsan", age=18, sex=SexEnum.FEMALE.value)

SQL語(yǔ)句

  • User實(shí)體類
from sqlmodel import SQLModel, Field, BigInteger, String, Integer

class User(SQLModel, table=True):
    id: int | None = Field(
        sa_type=BigInteger,
        default=None,
        primary_key=True,
        sa_column_kwargs={"comment": "自增id"},
    )
    name: str = Field(sa_type=String, sa_column_kwargs={"comment": "用戶名"})
    age: int = Field(sa_type=Integer, default=0, sa_column_kwargs={"comment": "年齡"})

sql語(yǔ)句

  • 執(zhí)行語(yǔ)句中的
from sqlmodel import Session, text

# 使用sql語(yǔ)句
def sql():
    with Session(engine) as session:
        statement = text('SELECT * FROM "user";')
        result = session.exec(statement).all()  # .all()表示多個(gè)返回值
        return result

# 帶參數(shù)
def sql_with_params():
    with Session(engine) as session:
        statement = text('SELECT * FROM "user" WHERE id=:id;')
        result = session.exec(statement, params={"id": 2}).one()    # .one()表示單個(gè)返回值
        return result


session.exec(statement).all()           # 返回多個(gè)值
session.exec(statement).one()           # 返回單一值（返回0個(gè)或多個(gè)都會(huì)報(bào)錯(cuò)）
session.exec(statement).one_or_none()   # 返回一個(gè)或None
session.exec(statement).first()         # 返回一個(gè)或None(性能不好，語(yǔ)句是查詢所有，然后返回第一條)
  • 下面的語(yǔ)句實(shí)際都可以轉(zhuǎn)換成sql
def to_sql():
    with Session(engine) as session:
        statement = select(User)
        
        print(statement)
        # SELECT "user".id, "user".name, "user".age FROM "user"
        
        result = session.exec(statement).all()
        return result

CRUD

新增

  • 最佳實(shí)踐
from sqlmodel import Session

def insert_entity(entity: User) -> None:
    with Session(engine) as session:
        session.add(entity)
        session.commit()
        # 重新綁定實(shí)體
        session.refresh(entity)


user = User(name="張三", age=18)
insert_entity(user)
print(user) # 可獲取完整數(shù)據(jù)，包括id
  • 其他寫(xiě)法
from sqlmodel import Session, insert

def insert_entity_1() -> None:
    with Session(engine) as session:
        statement = insert(User).values(name="李四", age=10)
        session.exec(statement)
        session.commit()

def insert_entity_2() -> None:
    with Session(engine) as session:
        statement = insert(User).values({"name": "王五", "age": 20})
        session.exec(statement)
        session.commit()

def insert_entity_3(entity: User) -> None:
    with Session(engine) as session:
        statement = insert(User).values(entity.model_dump(exclude_none=True))   # 使用pydantic轉(zhuǎn)換成字典(排除None是避免id傳入None報(bào)錯(cuò))
        session.exec(statement)
        session.commit()

更新

  • 最佳實(shí)踐
from sqlmodel import Session


def update_entity(entity: User) -> User:
    with Session(engine) as session:
        db_entity = session.get(User, entity.id)
        db_entity.sqlmodel_update(entity.model_dump())
        session.add(db_entity)
        session.commit()
        # 重新綁定實(shí)體
        session.refresh(db_entity)
        return db_entity


user = User(id=3, name="李四", age=20)
update_entity(user)
  • 其他更新語(yǔ)句
    • 無(wú)法返回更新后真正結(jié)果
from sqlmodel import Session, update

# 配合where語(yǔ)句可用于批量更新
def update_with_args() -> None:
    with Session(engine) as session:
        statement = update(User).where(User.id == 2).values(name="args")
        session.exec(statement)
        session.commit()

def update_with_kwargs() -> None:
    with Session(engine) as session:
        statement = update(User).where(User.id == 2).values({"name": "test"})
        session.exec(statement)
        session.commit()

def update_entity(user: User) -> None:
    with Session(engine) as session:
        statement = update(User).where(User.id == user.id).values(user.model_dump())
        session.exec(statement)
        session.commit()

刪除

  • 最佳實(shí)踐
from sqlmodel import Session, delete


def delete_by_id(id: int) -> None:
    with Session(engine) as session:
        statement = delete(User).where(User.id == id)
        session.exec(statement)
        session.commit()
  • 其他用法
from sqlmodel import Session


def delete_by_id(id: int) -> None:
    with Session(engine) as session:
        db_entity = session.get(User, id)
        session.delete(db_entity)
        session.commit()

簡(jiǎn)單查詢

  • 使用id查詢
from sqlmodel import Session

def select_by_id(id: int) -> User | None:
    with Session(engine) as session:
        user = session.get(User, id)
        return user
  • 使用where語(yǔ)句
from sqlmodel import Session, select

# 根據(jù)條件查詢
def select_by_where(name: str, age: int) -> list[User]:
    with Session(engine) as session:
        statement = select(User).where(User.name == name, User.age == age)
        result = session.exec(statement).all()
        return result

# 根據(jù)可選條件查詢
def select_by_condition(name: str | None = None, age: int | None = None) -> list[User]:
    with Session(engine) as session:
        statement = select(User)

        if name is not None:
            statement = statement.where(User.name == name)
        if age is not None:
            statement = statement.where(User.age == age)

        result = session.exec(statement).all()
        return result
  • like
from sqlmodel import Session, select

# like語(yǔ)句
def select_by_like(name: str) -> list[User]:
    with Session(engine) as session:
        statement = select(User).where(User.name.like(f"%{name}%"))
        result = session.exec(statement).all()
        return result

復(fù)雜查詢

  • 獲取數(shù)量
from sqlmodel import Session, select, func

# 數(shù)量
def select_count():
    with Session(engine) as session:
        statement = select(func.count()).select_from(User)
        # 增加篩選
        # statement = statement.where(User.age > 18)
        result = session.exec(statement).one()
        return result
  • 分頁(yè)
from sqlmodel import Session, select


def select_page() -> list[User]:
    with Session(engine) as session:
        statement = select(User).offset(0).limit(3)
        result = session.exec(statement).all()
        return result


# 分頁(yè)查詢包括總數(shù)量
def select_page_with_count(
    offset: int = 0, limit: int = 10, name: str | None = None, age: int | None = None
):
    with Session(engine) as session:
        conditions = []
        if name is not None:
            conditions.append(User.name.like(f"%{name}%"))

        if age is not None:
            conditions.append(User.age == age)

        # 拼接條件(第一個(gè)參數(shù)True作用是避免條件為空)
        whereclause = and_(True, *conditions)

        # 語(yǔ)句
        count_statement = select(func.count()).select_from(User).where(whereclause)
        list_statement = select(User).where(whereclause).offset(offset).limit(limit)
        return {
            "count": session.exec(count_statement).one(),
            "list": session.exec(list_statement).all(),
        }
  • 排序
from sqlmodel import Session, select, asc, desc

# 數(shù)量
def select_by_order():
    with Session(engine) as session:
        statement = select(User).order_by(desc(User.age))
        result = session.exec(statement).all()
        return result

FastAPI

  • response_model中如果設(shè)置了UserPageResp
  • data的list[Any]會(huì)自動(dòng)轉(zhuǎn)成list[UserItem]
# 分頁(yè)響應(yīng)表格結(jié)果
class TablePageResp(PageResp):
    total: int = Field(default=0, title="總數(shù)量")
    data: list[Any] = Field(default=[], title="數(shù)據(jù)")

# 響應(yīng)結(jié)果
class UserPageResp(PageResp):
    total: int = Field(default=0, title="總數(shù)量")
    data: list[UserItem] = Field(default=[], title="數(shù)據(jù)")
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容