Map RepositoryRepositoryDependencyAssociation declaratively

Also, setup Base class for declarative
This commit is contained in:
Sergey Golitsynskiy
2021-10-20 13:06:49 -04:00
parent 8f79a8fc46
commit 86a30bc0ff
3 changed files with 36 additions and 20 deletions
+34 -4
View File
@@ -3,8 +3,20 @@ import os
from enum import Enum
from typing import TYPE_CHECKING
from sqlalchemy.orm import registry
from sqlalchemy import (
Column,
DateTime,
ForeignKey,
Integer,
Table,
)
from sqlalchemy.orm import (
registry,
relationship,
)
from sqlalchemy.orm.decl_api import DeclarativeMeta
from galaxy.model.orm.now import now
from galaxy.util import asbool
from galaxy.util.bunch import Bunch
from galaxy.util.dictifiable import Dictifiable
@@ -15,14 +27,23 @@ log = logging.getLogger(__name__)
mapper_registry = registry()
if TYPE_CHECKING:
from sqlalchemy.schema import Table
class _HasTable:
table: Table
else:
_HasTable = object
class Base(metaclass=DeclarativeMeta):
__abstract__ = True
registry = mapper_registry
metadata = mapper_registry.metadata
__init__ = mapper_registry.constructor
@classmethod
def __declare_last__(cls):
cls.table = cls.__table__
class ToolShedRepository(_HasTable):
dict_collection_visible_keys = ['id', 'tool_shed', 'name', 'owner', 'installed_changeset_revision', 'changeset_revision', 'ctx_rev', 'includes_datatypes',
'tool_shed_status', 'deleted', 'uninstalled', 'dist_to_shed', 'status', 'error_message', 'description']
@@ -487,7 +508,16 @@ class ToolShedRepository(_HasTable):
return False
class RepositoryRepositoryDependencyAssociation(_HasTable):
class RepositoryRepositoryDependencyAssociation(Base, _HasTable):
__tablename__ = 'repository_repository_dependency_association'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
tool_shed_repository_id = Column(Integer, ForeignKey('tool_shed_repository.id'), index=True)
repository_dependency_id = Column(Integer, ForeignKey('repository_dependency.id'), index=True)
repository = relationship('ToolShedRepository')
repository_dependency = relationship('RepositoryDependency')
def __init__(self, tool_shed_repository_id=None, repository_dependency_id=None):
self.tool_shed_repository_id = tool_shed_repository_id
+1 -14
View File
@@ -42,13 +42,6 @@ install_model.ToolShedRepository.table = Table("tool_shed_repository", metadata,
Column("status", TrimmedString(255)),
Column("error_message", TEXT))
install_model.RepositoryRepositoryDependencyAssociation.table = Table('repository_repository_dependency_association', metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
Column("update_time", DateTime, default=now, onupdate=now),
Column("tool_shed_repository_id", Integer, ForeignKey("tool_shed_repository.id"), index=True),
Column("repository_dependency_id", Integer, ForeignKey("repository_dependency.id"), index=True))
install_model.RepositoryDependency.table = Table("repository_dependency", metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
@@ -87,13 +80,7 @@ mapper_registry.map_imperatively(install_model.ToolShedRepository, install_model
order_by=install_model.ToolDependency.table.c.name,
backref='tool_shed_repository'),
required_repositories=relation(install_model.RepositoryRepositoryDependencyAssociation,
primaryjoin=(install_model.ToolShedRepository.table.c.id == install_model.RepositoryRepositoryDependencyAssociation.table.c.tool_shed_repository_id))))
mapper_registry.map_imperatively(install_model.RepositoryRepositoryDependencyAssociation, install_model.RepositoryRepositoryDependencyAssociation.table,
properties=dict(repository=relation(install_model.ToolShedRepository,
primaryjoin=(install_model.RepositoryRepositoryDependencyAssociation.table.c.tool_shed_repository_id == install_model.ToolShedRepository.table.c.id)),
repository_dependency=relation(install_model.RepositoryDependency,
primaryjoin=(install_model.RepositoryRepositoryDependencyAssociation.table.c.repository_dependency_id == install_model.RepositoryDependency.table.c.id))))
primaryjoin=(install_model.ToolShedRepository.table.c.id == install_model.RepositoryRepositoryDependencyAssociation.tool_shed_repository_id))))
mapper_registry.map_imperatively(install_model.RepositoryDependency, install_model.RepositoryDependency.table,
properties=dict(repository=relation(install_model.ToolShedRepository,
@@ -111,8 +111,7 @@ class TestToolShedRepository(BaseTest):
class TestRepositoryRepositoryDependencyAssociation(BaseTest):
def test_table(self, cls_):
assert cls_.table.name == 'repository_repository_dependency_association'
# assert cls_.__tablename__ == 'repository_dependency' # TODO
assert cls_.__tablename__ == 'repository_repository_dependency_association'
def test_columns(self, session, cls_, repository, repository_dependency):
create_time = datetime.now()