Search code examples
pythonsqlalchemyfastapi

Query a SQLModel class with abstract definition


In this project, we have restricted imports between packages as follows:

  • We have a shared package.
  • We have a backend package.
  • Imports from the backend package to the shared package are not allowed, but imports from the shared package to the backend package are permitted. I cannot change this config.
  • Currently, some backend models, specifically SQLModel classes, need to be used in the shared package.

Here’s the solution I have implemented (paths are shortened):

in shared/models.py:

class SharedBaseSQLModel(SQLModel):
    # some shared fields like id
    # customized methods like save and read for shared package


class Job(SharedBaseSQLModel): # without table=True
     __abstract__ = True
     # some fileds like name, status

in backend/models.py:

from sqlmodel import SQLModel
from shared.models import Job as SharedJob


class BaseSQLModel(SQLModel):
   # customized methods like save and read for backend


class Job(BaseSQLModel, SharedJob, table=True): # here I use table=True
    # some extra methods

In shared I need to be able to do this query:

statement = select(Job).where(Job.name == 'foo').order_by(...).limit(...)
return session.exec(statement).all()

Error happens here: select(Job)

  • Current error: sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns clause element expected, got <class '..._shared.core.model.Job'>

  • Python 3.10.14

  • sqlalchemy-spanner==1.7.0

  • sqlmodel==0.0.18

As far as I understood, If I cannot use table=True, I cannot use select(MyModel).

Do you have any suggestions on how to solve this problem? I'm also open to ideas for a better model design to handle situations like this.


Solution

  • So, I could find a working solution.

    in shared/models.py:

    class SharedBaseSQLModel(SQLModel):
        # some shared fields like id
        # customized methods like save and read for shared package
    
    
    class Job(SharedBaseSQLModel, table=True): # with table=True <--
         __tablename__ = "job" # <--
         # some fileds like name, status
    

    in backend/models.py:

    from sqlmodel import SQLModel
    from shared.models import Job as SharedJob
    
    
    class BaseSQLModel(SQLModel):
       # customized methods like save and read for backend
    
    
    class Job(BaseSQLModel, SharedJob, table=False): # with table=False <--
        # some extra methods
        __abstract__ = True # <--
    
    

    But the main change was added here foo/bar/main.py:

    from sqlalchemy import MetaData
    from sqlalchemy.orm import registry
    
    from shared.models import SharedBaseSQLModel
    
    from backend.models import BaseSQLModel
    from backend.core.models.job import Job # we use backend model here
    
    combined_metadata = MetaData()
    
    # pay attention to order first shared then backend
    shared_metadata = SharedBaseSQLModel.metadata
    backend_metadata = BaseSQLModel.metadata
    
    for table in shared_metadata.tables.values():
        table.tometadata(combined_metadata)
    
    for table in backend_metadata.tables.values():
        table.tometadata(combined_metadata)
    
    mapper_registry = registry(metadata=combined_metadata)
    mapper_registry.map_imperatively(Job, Job.__table__) # this __table__ comes from shared!
    
    combined_metadata.create_all(engine) # you need to import engine
    

    In this way the CRUD operations are accessible from in shared and backend. This definition should solve the sqlalchemy.orm.exc.UnmappedClassError and sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns clause element expected, got <class 'foo.bar.model.YourModel'

    UPDATE: There is another solution:

    in shared/models.py:

    class SharedBaseSQLModel(SQLModel):
        # some shared fields like id
        # customized methods like save and read for shared package
    
    
    class BaseJob(SharedBaseSQLModel): # without table=True/False <--
         # some fileds like name, status
    

    in backend/models.py:

    from sqlmodel import SQLModel
    from shared.models import import BaseJob
    
    
    class BaseSQLModel(SQLModel):
       # customized methods like save and read for backend
    
    
    class Job(BaseJob, BaseSQLModel, table=True): # with table=True <--
        __table_args__ = {"extend_existing": True} # <-- 
        # when table=True everywhere that we need this model, 
        # we should add this line, apart from BaseJob
        # some extra methods
    
    

    foo/bar/main.py

    SharedBaseSQLModel.metadata.create_all(engine)