如何在SQLModel中创建SQL视图？附Hero与Team关联表示例

SQLModel本身未提供直接定义视图的API，但它基于SQLAlchemy构建，因此可以借助SQLAlchemy的CreateView对象实现视图创建。以下是针对你的Hero和Team表，创建展示英雄姓名及对应团队名称视图的具体步骤：

1. 定义视图对应的模型类 #

先定义一个与视图结构匹配的SQLModel类，无需设置table=True（它对应数据库视图而非物理表）：

from sqlmodel import SQLModel, Field

class HeroTeamView(SQLModel):
    hero_name: str = Field(alias="name")
    team_name: str | None = Field(alias="team_name")

    class Config:
        orm_mode = True

2. 构建视图SQL并创建视图 #

导入SQLAlchemy工具后，在数据库初始化流程中添加视图创建逻辑：

from sqlalchemy import text
from sqlalchemy.sql.schema import CreateView

def create_db_and_tables():
    # 先创建物理表
    SQLModel.metadata.create_all(engine)
    
    # 定义视图查询语句，用LEFT JOIN保留无团队的英雄
    view_sql = text("""
        SELECT 
            hero.name,
            team.name AS team_name
        FROM hero
        LEFT JOIN team ON hero.team_id = team.id
    """)
    
    # 创建视图对象并绑定到metadata
    hero_team_view = CreateView(
        "hero_team_view",
        view_sql,
        metadata=SQLModel.metadata
    )
    
    # 执行视图创建
    with engine.connect() as conn:
        conn.execute(hero_team_view)
        conn.commit()

3. 查询视图数据 #

视图创建完成后，可像查询普通模型一样查询视图：

def query_hero_team_view():
    with Session(engine) as session:
        hero_teams = session.query(HeroTeamView).all()
        print("\n英雄与团队对应关系：")
        for item in hero_teams:
            print(f"英雄: {item.hero_name}, 所属团队: {item.team_name or '无'}")

完整修改后的代码 #

将上述逻辑整合到原代码中，完整代码如下：

from sqlmodel import Field, Session, SQLModel, create_engine
from sqlalchemy import text
from sqlalchemy.sql.schema import CreateView


class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

    team_id: int | None = Field(default=None, foreign_key="team.id")


class HeroTeamView(SQLModel):
    hero_name: str = Field(alias="name")
    team_name: str | None = Field(alias="team_name")

    class Config:
        orm_mode = True


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
    
    view_sql = text("""
        SELECT 
            hero.name,
            team.name AS team_name
        FROM hero
        LEFT JOIN team ON hero.team_id = team.id
    """)
    
    hero_team_view = CreateView(
        "hero_team_view",
        view_sql,
        metadata=SQLModel.metadata
    )
    
    with engine.connect() as conn:
        conn.execute(hero_team_view)
        conn.commit()


def create_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret's Bar")
        session.add(team_preventers)
        session.add(team_z_force)
        session.commit()

        hero_deadpond = Hero(
            name="Deadpond", secret_name="Dive Wilson", team_id=team_z_force.id
        )
        hero_rusty_man = Hero(
            name="Rusty-Man",
            secret_name="Tommy Sharp",
            age=48,
            team_id=team_preventers.id,
        )
        hero_spider_boy = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
        session.add(hero_deadpond)
        session.add(hero_rusty_man)
        session.add(hero_spider_boy)
        session.commit()

        session.refresh(hero_deadpond)
        session.refresh(hero_rusty_man)
        session.refresh(hero_spider_boy)

        print("Created hero:", hero_deadpond)
        print("Created hero:", hero_rusty_man)
        print("Created hero:", hero_spider_boy)


def query_hero_team_view():
    with Session(engine) as session:
        hero_teams = session.query(HeroTeamView).all()
        print("\n英雄与团队对应关系：")
        for item in hero_teams:
            print(f"英雄: {item.hero_name}, 所属团队: {item.team_name or '无'}")


def main():
    create_db_and_tables()
    create_heroes()
    query_hero_team_view()


if __name__ == "__main__":
    main()

关键说明 #

• 视图创建需在物理表之后执行，避免表不存在的错误。
• HeroTeamView类通过alias属性映射视图查询字段，orm_mode=True确保SQLModel能将查询结果转为模型实例。
• 使用LEFT JOIN保证无团队的英雄也能被查询到，符合业务需求。

内容的提问来源于stack exchange，提问作者robertspierre