import re from sqlalchemy.ext.declarative import ( declarative_base, declared_attr, ) from sqlalchemy import ( MetaData, Index, ) from . import ( commit_session, get_session, ) INDEX_CONVENTION = { "ix": "ix_%(table_name)s_%(column_0_label)s", "uq": "uq_%(table_name)s_%(column_0_name)s", "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", "pk": "pk_%(table_name)s" } METADATA = MetaData(naming_convention=INDEX_CONVENTION) Base = declarative_base(metadata=METADATA) def class_name_to_underscores(name): """Helper to turn a class name in camelCase to camel_case""" s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower() class ModelMixin: """Mixin to give models some helpful functionality""" @declared_attr def __tablename__(cls): """Automatically create table names using a convention from the name of the model class""" name = class_name_to_underscores(cls.__name__) if name.endswith('y'): name = name.rstrip('y') + 'ies' elif not name.endswith('s'): name = name + 's' return name def save(self, *, commit=True): """Add in a standard way of saving a model instance""" session = get_session() session.add(self) if commit: commit_session() @classmethod def flush(self): # pragma: no cover get_session().flush() @classmethod def _get_index(cls, name, *columns, **kwargs): # pragma: no cover cols = [getattr(cls, c) for c in columns] return Index(name, *cols, **kwargs) @classmethod def generate_index(cls, *columns, **kwargs): # pragma: no cover name_mapping = { 'table_name': cls.__tablename__, 'column_0_label': '_'.join(columns), } name = INDEX_CONVENTION['ix'] % name_mapping return cls._get_index(name, *columns, **kwargs) @classmethod def generate_unique_index(cls, *columns, **kwargs): # pragma: no cover name_mapping = { 'table_name': cls.__tablename__, 'column_0_name': '_'.join(columns), } name = INDEX_CONVENTION['uq'] % name_mapping if not kwargs.get('unique'): kwargs['unique'] = True return cls._get_index(name, *columns, **kwargs)