I need to set different __table_args__ for the mapped class depending on the backend DBMS, e.g. mssql and mysql - and this attribute cannot be set after the class definition, e.g. class Repository(Base) has been evaluated.

Ideally, my db/models.py module would take a parameter db_type, when it's imported, but this is not possible with Python.

For the module to export the local (mapped) classes, I use the following code:

tables = (Repository, Commit, Blob)
globals().update((t.__name__, t) for t in tables)

But with this approach, it's not possible for importing modules to write:

from db.models import Repository, Commit, Blob

since the classes are not exported at that point, but only after create_models() has been called.

What's the best approach for setting these __table_args__ in a way that other modules can import the mapped classes like above, and so that I avoid having to have local class definitions in a function?


import sqlalchemy
from sqlalchemy import (
    Table, Column, ForeignKey, Integer,
    String, Text, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import ForeignKeyConstraint
from sqlalchemy.orm import relationship

from common.exc import WatchdogException

_Base = None

def create_models(db_type):
    global _Base

    if _Base:
        return _Base

    if db_type == 'mssql':
        table_args = {}
    elif db_type == 'mysql':
        table_args = {'mysql_default_charset': 'utf8',
                      'mysql_collate': 'utf8_bin'}
        raise WatchdogException('The database type is not supported: {}'.format(db_type))

    Base = declarative_base()

    class Repository(Base):
        __tablename__ = 'stash_repository'

        id = Column(Integer, primary_key=True)
        head_branch = Column(Text, nullable=False)
        head_commit = Column(String(40), nullable=False)
        slug = Column(Text, nullable=False)
        name = Column(Text, nullable=False)
        state = Column(Text, nullable=False)
        href = Column(Text, nullable=False)
        clone = Column(Text, nullable=False)

        commits = relationship('Commit', back_populates='repository', cascade='all, delete-orphan')  # post_update=True

        __table_args__ = (
            ForeignKeyConstraint([project_id], [Project.id], onupdate='CASCADE', ondelete='CASCADE'),  # use_alter=True

        def __repr__(self):
            return f'<{self.__class__.__name__}(id=\'{self.id}\')>'

    class Commit(Base):
        __tablename__ = 'stash_commit'

        id = Column(String(40), primary_key=True)
        repository_id = Column(Integer, primary_key=True)
        author_name = Column(Text, nullable=False)
        author_email = Column(Text, nullable=False)
        author_time = Column(DateTime, nullable=False)
        committer_name = Column(Text, nullable=False)
        committer_email = Column(Text, nullable=False)
        commit_time = Column(DateTime, nullable=False)
        message = Column(Text, nullable=False)

        repository = relationship('Repository', back_populates='commits')
        blobs = relationship('Blob', back_populates='commit', cascade='all, delete-orphan')

        __table_args__ = (
            ForeignKeyConstraint([repository_id], [Repository.id], onupdate='CASCADE', ondelete='CASCADE'),

        def __repr__(self):
            return f'<{self.__class__.__name__}(id=\'{self.id}\')>'

    class Blob(Base):
        __tablename__ = 'stash_blob'

        id = Column(String(40), primary_key=True)
        commit_id = Column(String(40), primary_key=True)
        name = Column(Text, nullable=False)
        path = Column(Text, nullable=False)
        mime = Column(Text, nullable=False)
        infer_mime = Column(Text, nullable=True)
        size = Column(Integer, nullable=False)

        commit = relationship('Commit', back_populates='blobs')
        matches = relationship('Match', back_populates='blob')

        __table_args__ = (
            ForeignKeyConstraint([commit_id], [Commit.id], onupdate='CASCADE', ondelete='CASCADE'),

        def __repr__(self):
            return f'<{self.__class__.__name__}(id=\'{self.id}\')>'

    tables = (Repository, Commit, Blob)
    globals().update((t.__name__, t) for t in tables)

    _Base = Base
    return _Base
  • \$\begingroup\$ You define a mixin (or a base) and the models will derive from that. \$\endgroup\$ – hjpotter92 Aug 21 '20 at 4:37
  • \$\begingroup\$ @hjpotter92 - Can you some sample code? A mixin is just a fancy term for a base class, right, when multiple base classes are used? Would the base class set the __table_args__? Also, this still doesn't solve my issue of having to create the classes dynamically in a function, does it? \$\endgroup\$ – user229443 Aug 21 '20 at 8:24
  • \$\begingroup\$ What's refraining you from having from db.models import create_models; Repository, Commit, Blob = create_models('mysql')? Provided you return the right values. \$\endgroup\$ – 301_Moved_Permanently Aug 23 '20 at 11:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.