SQLAlchemy-Continuum compatibility

[petyunchik]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Optional

from sqlalchemy_continuum import make_versioned
from sqlmodel import Field, Session, SQLModel, create_engine
from sqlmodel.main import default_registry


# setattr(SQLModel, 'registry', default_registry)
make_versioned(user_cls=None)


class Hero(SQLModel, table=True):
    __versioned__ = {}

    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")

engine = create_engine("sqlite:///database.db")


SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    session.add(hero_1)
    session.commit()
    session.refresh(hero_1)
    print(hero_1)

Description

Very basic setup of SQLAlchemy-Continuum doesn't work with SQLModel. Following error raised (see here):

AttributeError: type object 'SQLModel' has no attribute '_decl_class_registry'

SQLModel class misses some expected attribute (registry for SQLAlchemy >= 1.4) from SQLAlchemy's Base. For debugging purposes the attribute can be manually populated (uncomment # setattr(SQLModel, 'registry', default_registry) in the example code) and it helps to proceed to the next error:

  <skipped>
  File ".../.venv/lib/python3.10/site-packages/sqlmodel/main.py", line 277, in __new__
    new_cls = super().__new__(cls, name, bases, dict_used, **config_kwargs)
  File "pydantic/main.py", line 228, in pydantic.main.ModelMetaclass.__new__
  File "pydantic/fields.py", line 488, in pydantic.fields.ModelField.infer
  File "pydantic/fields.py", line 419, in pydantic.fields.ModelField.__init__
  File "pydantic/fields.py", line 528, in pydantic.fields.ModelField.prepare
  File "pydantic/fields.py", line 552, in pydantic.fields.ModelField._set_default_and_type
  File "pydantic/fields.py", line 422, in pydantic.fields.ModelField.get_default
  File "pydantic/utils.py", line 652, in pydantic.utils.smart_deepcopy
  File ".../.venv/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 582, in __bool__
    raise TypeError("Boolean value of this clause is not defined")
TypeError: Boolean value of this clause is not defined

Looks like there is some incompatibility with SQLAlchemy's expected behaviour, because this error was thrown just after SQLModel's __new__ method call.

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.6

Python Version

3.10.2

Additional Context

No response

[coneybeare]

For what it's worth, I am seeing this error even without sqlmodel or sqlalchemy_continuum. Did you ever find the root cause?

[igkins]

@petyunchik were you able to get this working? upgrading pydantic fixes that issue, however, there are a couple other issues that then you have to work through, both with continuum and sqlmodel

[ftapajos]

As of today (i.e. sqlmodel 0.0.14 and sqlalchemy-continuum 1.4.0), the error is

pydantic.errors.PydanticUserError: A non-annotated attribute was detected: `id = Column(None, BigInteger(), table=None, primary_key=True, nullable=False, default=Sequence('transaction_id_seq'))`. All model fields require a type annotation; if `id` is not meant to be a field, you may be able to resolve this error by annotating it as a `ClassVar` or updating `model_config['ignored_types']`.

Couldn't debug it properly, but I assume this is due to the fact that continuum create sqlalchemy models for versioning, but do not annotate these models. I think I'll have to deal with this problem, because versioning is of the essence for some of my applications. My intuition tells that it is best to create a new module by patching sqlalchemy-continuum to use sqlmodel's Base and registry, rather than adapting any of these to use the other

[abrichr]

@ftapajos thank you for the suggestion. Any update on this? I am interested in contributing 🙏

[abrichr]

Related: kvesteri/sqlalchemy-continuum#271

@tiangolo any thoughts on this would be greatly appreciated! 🙏

[ftapajos]

@abrichr I haven't even started yet, but I could follow your lead

[CiberNin]

Really hoping a solution is in the works here.

[AlePiccin]

Can't wait to be compatible. I really need this feature.

[AlePiccin]

I found a solution. I'm currently using:

fastapi[standard]==0.115.5
pydantic==2.8.2
pydantic-settings==2.2.1
sqlmodel==0.0.22
sqlalchemy==2.0.36
sqlalchemy-utils==0.41.2
sqlalchemy-continuum==1.4.2

Before declaring any model, do the following:

import sqlalchemy_utils

def new_get_declarative_base(model):
    import sqlmodel
    if issubclass(model, sqlmodel.SQLModel):
        return model._sa_registry.generate_base()
    else:
        for parent in model.__bases__:
            try:
                _ = parent.metadata
                return new_get_declarative_base(parent)
            except AttributeError:
                pass
        return model


sqlalchemy_utils.functions.get_declarative_base = new_get_declarative_base


import sqlmodel

sqlmodel.SQLModel.metadata = sqlmodel.main.default_registry.metadata

sqlalchemy_continuum.make_versioned()

class BaseSQLModel(sqlmodel.SQLModel):
    __versioned__ = {}

This should work with SQLModels. Your models should inherit from BaseSQLModel.

If you use SQL Server as your database, you may encounter an error due to a sequence definition in the id column of the TransactionClass. You can rewrite the TransactionClass as shown below to address this issue. Additionally, I use a FastAPIPlugin to retrieve the user from the session. All you need to do is set the user data in the session.info dictionary. Here is the complete code I use:

import sqlalchemy_utils


def new_get_declarative_base(model):
    import sqlmodel
    if issubclass(model, sqlmodel.SQLModel):
        # noinspection PyProtectedMember
        return model._sa_registry.generate_base()
    else:
        for parent in model.__bases__:
            try:
                _ = parent.metadata
                return new_get_declarative_base(parent)
            except AttributeError:
                pass
        return model


sqlalchemy_utils.functions.get_declarative_base = new_get_declarative_base

import sqlalchemy_continuum


def new_create_class(self, manager):
    """
    Create Transaction class.
    """
    import sqlalchemy as sa
    from collections import OrderedDict

    class Transaction(manager.declarative_base, sqlalchemy_continuum.transaction.TransactionBase):
        __tablename__ = 'transaction'
        __versioning_manager__ = manager

        id = sa.Column(sa.types.BigInteger, primary_key=True, autoincrement=True)

        if self.remote_addr:
            remote_addr = sa.Column(sa.String(50))

        if manager.user_cls:
            user_cls = manager.user_cls
            Base = manager.declarative_base
            # noinspection PyProtectedMember
            registry = Base.registry._class_registry

            if isinstance(user_cls, str):
                try:
                    user_cls = registry[user_cls]
                except KeyError:
                    raise sqlalchemy_continuum.ImproperlyConfigured('Could not build relationship between Transaction'
                                                                    ' and %s. %s was not found in declarative class '
                                                                    'registry. Either configure VersioningManager to '
                                                                    'use different user class or disable this '
                                                                    'relationship ' % (user_cls, user_cls))

            user_id = sa.Column(sa.inspect(user_cls).primary_key[0].type,
                sa.ForeignKey(sa.inspect(user_cls).primary_key[0]), index=True)

            user = sa.orm.relationship(user_cls)

        def __repr__(self):
            fields = ['id', 'issued_at', 'user']
            field_values = OrderedDict((field, getattr(self, field)) for field in fields if hasattr(self, field))
            return '<Transaction %s>' % ', '.join(('%s=%r' % (field, value) if not isinstance(value,
                                                                                              int) # We want the following line to ensure that longs get
            # shown without the ugly L suffix on python 2.x
            # versions
            else '%s=%d' % (field, value) for field, value in field_values.items()))

    if manager.options['native_versioning']:
        sqlalchemy_continuum.transaction.create_triggers(Transaction)
    return Transaction


sqlalchemy_continuum.transaction.TransactionFactory.create_class = new_create_class

from sqlalchemy_continuum.plugins import Plugin


class FastAPIPlugin(Plugin):

    def transaction_args(self, uow, session):
        return {'user_id': session.info.get('id_usuario'), 'remote_addr': None, }


import sqlmodel
from sqlalchemy.orm import declared_attr

constraint_naming_conventions = {"ix": "ix_%(column_0_label)s", "uq": "uq_%(table_name)s_%(column_0_name)s",
                                 "ck": "ck_%(table_name)s_%(constraint_name)s",
                                 "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
                                 "pk": "pk_%(table_name)s"}

sqlmodel.main.default_registry.metadata = sqlmodel.MetaData(naming_convention=constraint_naming_conventions)

sqlmodel.SQLModel.metadata = sqlmodel.main.default_registry.metadata

sqlalchemy_continuum.make_versioned(user_cls="Usuario", plugins=[FastAPIPlugin()])


class BaseSQLModel(sqlmodel.SQLModel):
    __versioned__ = {}

    @declared_attr  # type: ignore
    def __tablename__(cls) -> str:
        return cls.__name__