FastAPI+SQLAlchemy无外键多对多模型异步查询中间表扩展字段

一、模型修正与关联关系建立 #

首先，SQLAlchemy模型需继承自SQLAlchemy的Base（而非Pydantic的BaseModel），这是异步查询的前提。同时需要在DkCatalog和DkCatalogTable之间通过中间表建立关联，才能方便地关联查询中间表字段：

from sqlalchemy import Column, String, Integer, Boolean, DateTime, func, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.associationproxy import association_proxy

Base = declarative_base()

class DkCatalogTable(Base):
    __tablename__ = 'dk_catalog_table'
    # 原有字段保持不变
    datasource_id = Column("datasource_id", String(255), comment="数据源ID")
    datasource = relationship('DkDataSourceinfo',
                              primaryjoin='foreign(DkCatalogTable.datasource_id) == DkDataSourceinfo.id',
                              backref=backref('dk_catalog_table'))
    name = Column("name", String(128), comment="表中文名")
    physical_table_name = Column("physical_table_name", String(128), comment="物理表名")
    logic_table_name = Column("logic_table_name", String(128), comment="逻辑表名")
    table_name_alias = Column("table_name_alias", String(255), comment="表中文别名")
    table_code = Column("table_code", Integer, comment="表编码")
    fields = relationship('DkCatalogField',
                          primaryjoin='DkCatalogTable.table_code==foreign(DkCatalogField.table_code)',
                          backref=backref('dk_catalog_table'))
    table_type = Column("table_type", String(64), comment="表类型")
    is_open = Column("is_open", Boolean, default=True, comment="是否公开;1-公开;0不公开")
    is_edit = Column("is_edit", Boolean, default=False, comment="是否可编辑;1-可编辑;0不可编辑")
    order_no = Column("order_no", Integer, comment="排序")
    creator = Column("creator", String(255), comment="创建者")
    create_time = Column("create_time", DateTime, server_default=func.now(), comment="创建时间")
    last_modifier = Column("last_modifier", String(255), comment="上次修改用户")
    last_modify_time = Column("last_modify_time", DateTime, server_default=func.now(), onupdate=func.now(),
                              comment="上次修改时间")
    # 添加与中间表的关联
    catalog_relations = relationship("DkCatalogTableRelational", backref="table")
    # 通过association_proxy直接获取中间表字段
    catalog_order_no = association_proxy("catalog_relations", "order_no")
    catalog_is_show = association_proxy("catalog_relations", "is_show")

class DkCatalog(Base):
    __tablename__ = 'dk_catalog'
    # 原有字段保持不变
    name_cn = Column("name_cn", String(128), comment="中文名")
    name_en = Column("name_en", String(128), comment="英文名")
    catalog_code = Column("catalog_code", String(128), comment="目录编码")
    parent_id = Column("parent_id", String(255), comment="父级")
    parents = relationship("DkCatalog",
                           primaryjoin='DkCatalog.id==foreign(DkCatalog.parent_id)',
                           backref=backref('dk_catalog', remote_side='DkCatalog.id'))
    order_no = Column("order_no", Integer, comment="排序")
    is_show = Column("is_show", Boolean, default=True, comment="是否显示:1显示;0不显示")
    creator = Column("creator", String(255), comment="创建者")
    create_time = Column("create_time", DateTime, server_default=func.now(), comment="创建时间")
    last_modifier = Column("last_modifier", String(255), comment="上次修改用户")
    last_modify_time = Column("last_modify_time", DateTime, server_default=func.now(), onupdate=func.now(),
                              comment="上次修改时间")
    # 添加与中间表的关联
    table_relations = relationship("DkCatalogTableRelational", backref="catalog")
    # 通过association_proxy直接获取中间表字段
    table_order_no = association_proxy("table_relations", "order_no")
    table_is_show = association_proxy("table_relations", "is_show")

class DkCatalogTableRelational(Base):
    __tablename__ = 'dk_catalog_table_relational_info'
    id = Column(Integer, primary_key=True, autoincrement=True)  # 建议添加主键
    catalog_id = Column(String(128), ForeignKey("dk_catalog.id"), comment="目录编码ID")
    tabl_id = Column(String(128), ForeignKey("dk_catalog_table.id"), comment="表编码ID")
    # 原有字段保持不变
    order_no = Column("order_no", Integer, comment="排序")
    is_show = Column("is_show", Boolean, default=True, comment="是否显示:1显示;0不显示")
    creator = Column("creator", String(255), comment="创建者")
    create_time = Column("create_time", DateTime, server_default=func.now(), comment="创建时间")
    last_modifier = Column("last_modifier", String(255), comment="上次修改用户")
    last_modify_time = Column("last_modify_time", DateTime, server_default=func.now(), onupdate=func.now(),
                              comment="上次修改时间")

二、异步查询实现 #

先配置SQLAlchemy异步引擎与会话：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select

# 替换为你的数据库连接地址
engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname", echo=True)
async_session = AsyncSession(engine, expire_on_commit=False)

1. 直接查询中间表指定字段 #

仅需获取中间表的is_show和order_no时，可直接执行异步查询：

async def get_relational_fields(catalog_id: str = None):
    async with async_session() as session:
        query = select(DkCatalogTableRelational.order_no, DkCatalogTableRelational.is_show)
        if catalog_id:
            query = query.where(DkCatalogTableRelational.catalog_id == catalog_id)
        result = await session.execute(query)
        return result.all()

2. 关联查询目录/表时获取中间表字段 #

查询目录或表的同时带出中间表字段，利用已建立的关联关系：

查询目录及关联表的中间表字段

async def get_catalog_with_relational(catalog_id: str):
    async with async_session() as session:
        result = await session.execute(
            select(DkCatalog, DkCatalogTableRelational.order_no, DkCatalogTableRelational.is_show)
            .join(DkCatalog.table_relations)
            .where(DkCatalog.id == catalog_id)
        )
        return result.all()

查询表及关联目录的中间表字段

async def get_table_with_relational(table_id: str):
    async with async_session() as session:
        result = await session.execute(
            select(DkCatalogTable, DkCatalogTableRelational.order_no, DkCatalogTableRelational.is_show)
            .join(DkCatalogTable.catalog_relations)
            .where(DkCatalogTable.id == table_id)
        )
        return result.all()

三、FastAPI接口示例 #

将查询逻辑封装为接口：

from fastapi import FastAPI

app = FastAPI()

@app.get("/relational-fields/")
async def relational_fields_api(catalog_id: str = None):
    data = await get_relational_fields(catalog_id)
    return {"data": [{"order_no": item[0], "is_show": item[1]} for item in data]}

@app.get("/catalog/{catalog_id}/relational/")
async def catalog_relational_api(catalog_id: str):
    data = await get_catalog_with_relational(catalog_id)
    return {"data": [
        {
            "catalog_name": item[0].name_cn,
            "order_no": item[1],
            "is_show": item[2]
        } for item in data
    ]}

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