共享包使用抽象SQLModel类执行查询报错的解决方案求助

你遇到的报错本质是：shared包中的Job类是抽象模型（__abstract__ = True且未设置table=True），SQLAlchemy/SQLModel不会将抽象模型映射到数据库表，因此无法直接用它作为select()的查询目标。

方案1：动态导入后端实体类 #

既然不能在shared中直接导入backend包，可通过运行时动态导入的方式获取后端的Job实体类（真正绑定表的那个）：

import importlib

def get_jobs(session, name: str):
    # 动态导入backend的Job类
    try:
        backend_models = importlib.import_module("backend.models")
        BackendJob = backend_models.Job
    except (ImportError, AttributeError):
        raise ValueError("无法导入后端Job实体类")
    
    statement = select(BackendJob).where(BackendJob.name == name).order_by(...).limit(...)
    return session.exec(statement).all()

注意：需确保backend包在运行时的Python路径中可被找到，后续如果后端模型路径变更，要同步修改导入路径。

方案2：依赖注入传递实体类 #

让shared只定义通用查询逻辑，具体的实体类由调用方（backend）传入，完全规避跨包反向依赖：

# shared/queries.py
from sqlmodel import Session, select

def get_jobs_by_name(session: Session, job_model, name: str):
    statement = select(job_model).where(job_model.name == name).order_by(...).limit(...)
    return session.exec(statement).all()

然后在backend中调用：

# backend/services.py
from sqlmodel import Session
from shared.queries import get_jobs_by_name
from backend.models import Job

def fetch_jobs(session: Session, name: str):
    return get_jobs_by_name(session, Job, name)

这个方案更符合依赖倒置原则，逻辑清晰且易于维护。

思路1：分离数据模型与业务逻辑 #

shared包只定义**数据传输对象（DTO）**或纯字段结构，不涉及SQLModel实体；backend负责数据库实体定义与数据查询，通过接口或函数向shared提供数据：

# shared/dtos.py
from pydantic import BaseModel

class JobDTO(BaseModel):
    id: int
    name: str
    status: str
    # 其他需要在shared中使用的字段

# backend/services.py
from sqlmodel import Session, select
from backend.models import Job
from shared.dtos import JobDTO

def get_jobs_dto(session: Session, name: str) -> list[JobDTO]:
    db_jobs = session.exec(select(Job).where(Job.name == name)).all()
    return [JobDTO.model_validate(job) for job in db_jobs]

shared需要数据时，直接调用backend的服务函数，完全避免操作数据库实体。

思路2：用混合类替代抽象模型 #

将shared中的模型拆分为纯字段混合类，而非抽象SQLModel，让backend的实体类继承这些混合类来复用字段：

# shared/models.py
from sqlmodel import Field

class SharedBaseFields:
    id: int = Field(default=None, primary_key=True)
    # 其他共享字段

class JobFields:
    name: str
    status: str
    # 其他Job共享字段

# backend/models.py
from sqlmodel import SQLModel, table
from shared.models import SharedBaseFields, JobFields

class BaseSQLModel(SQLModel):
    # backend自定义方法

class Job(BaseSQLModel, SharedBaseFields, JobFields, table=True):
    # backend额外方法

这种方式下shared中没有可被误用于查询的模型类，字段复用更安全。

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