# -*- coding: utf-8 -*-
#
# Copyright (C) 2018 CERN.
#
# Asclepias Broker is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.
"""Graph database models."""
import enum
import uuid
from invenio_db import db
from sqlalchemy.schema import Index, PrimaryKeyConstraint, UniqueConstraint
from sqlalchemy_utils.models import Timestamp
from sqlalchemy_utils.types import UUIDType
from ..core.models import Identifier, Relation, Relationship
[docs]class GroupType(enum.Enum):
"""Group type."""
Identity = 1
Version = 2
[docs]class Group(db.Model, Timestamp):
"""Group model."""
__tablename__ = 'group'
id = db.Column(UUIDType, default=uuid.uuid4, primary_key=True)
type = db.Column(db.Enum(GroupType), nullable=False)
identifiers = db.relationship(
Identifier,
secondary=lambda: Identifier2Group.__table__,
backref='groups',
viewonly=True)
groups = db.relationship(
'Group',
secondary=lambda: GroupM2M.__table__,
primaryjoin=lambda: (Group.id == GroupM2M.group_id),
secondaryjoin=lambda: (Group.id == GroupM2M.subgroup_id))
def __repr__(self):
"""String representation of the group."""
return f"<{self.id}: {self.type.name}>"
[docs]class GroupRelationship(db.Model, Timestamp):
"""Group relationship model."""
__tablename__ = 'grouprelationship'
__table_args__ = (
UniqueConstraint('source_id', 'target_id', 'relation',
name='uq_grouprelationship_source_target_relation'),
# TODO: Change to "index=True"
Index('ix_grouprelationship_source', 'source_id'),
Index('ix_grouprelationship_target', 'target_id'),
Index('ix_grouprelationship_relation', 'relation'),
)
id = db.Column(UUIDType, default=uuid.uuid4, primary_key=True)
type = db.Column(db.Enum(GroupType), nullable=False)
relation = db.Column(db.Enum(Relation), nullable=False)
source_id = db.Column(
UUIDType,
db.ForeignKey(Group.id, ondelete='CASCADE', onupdate='CASCADE'),
nullable=False
)
target_id = db.Column(
UUIDType,
db.ForeignKey(Group.id, ondelete='CASCADE', onupdate='CASCADE'),
nullable=False
)
# DB relationships
source = db.relationship(
Group, foreign_keys=[source_id], backref='sources')
target = db.relationship(
Group, foreign_keys=[target_id], backref='targets')
relationships = db.relationship(
'GroupRelationship',
secondary=lambda: GroupRelationshipM2M.__table__,
primaryjoin=lambda: (GroupRelationship.id ==
GroupRelationshipM2M.relationship_id),
secondaryjoin=lambda: (GroupRelationship.id ==
GroupRelationshipM2M.subrelationship_id))
# TODO:
# We don't store 'deleted' as in the relation as most likely don't need
# that as 'ground truth' in precomputed groups anyway
def __repr__(self):
"""String representation of the group relationship."""
return f'<{self.source} {self.relation.name} {self.target}>'
[docs]class Identifier2Group(db.Model, Timestamp):
"""Many-to-many model for Identifier and Group."""
__tablename__ = 'identifier2group'
__table_args__ = (
PrimaryKeyConstraint('identifier_id', 'group_id',
name='pk_identifier2group'),
)
identifier_id = db.Column(
UUIDType,
db.ForeignKey(Identifier.id, ondelete='CASCADE', onupdate='CASCADE'),
nullable=False
)
group_id = db.Column(
UUIDType,
db.ForeignKey(Group.id, ondelete='CASCADE', onupdate='CASCADE'),
nullable=False
)
# DB relationships
identifier = db.relationship(
Identifier, foreign_keys=[identifier_id], backref='id2groups')
group = db.relationship(
Group, foreign_keys=[group_id], backref='id2groups')
[docs]class Relationship2GroupRelationship(db.Model, Timestamp):
"""Many-to-many model for Relationship to GroupRelationship."""
__tablename__ = 'relationship2grouprelationship'
__table_args__ = (
PrimaryKeyConstraint('relationship_id', 'group_relationship_id',
name='pk_relationship2grouprelationship'),
)
relationship_id = db.Column(
UUIDType,
db.ForeignKey(Relationship.id, onupdate='CASCADE', ondelete='CASCADE'),
nullable=False
)
group_relationship_id = db.Column(
UUIDType,
db.ForeignKey(
GroupRelationship.id, onupdate='CASCADE', ondelete='CASCADE'),
nullable=False)
# DB relationships
relationship = db.relationship(
Relationship, foreign_keys=[relationship_id],
backref='relationship2group_relationship')
group_relationship = db.relationship(
GroupRelationship, foreign_keys=[group_relationship_id],
backref='relationship2group_relationship')
def __repr__(self):
"""String representation of the model."""
return f'<{self.group_relationship}: {self.relationship}>'
[docs]class GroupM2M(db.Model, Timestamp):
"""Many-to-many model for Groups."""
__tablename__ = 'groupm2m'
__table_args__ = (
PrimaryKeyConstraint('group_id', 'subgroup_id', name='pk_groupm2m'),
UniqueConstraint('subgroup_id', name='uq_groupm2m_subgroup_id'),
)
group_id = db.Column(
UUIDType,
db.ForeignKey(Group.id, onupdate='CASCADE', ondelete='CASCADE'),
nullable=False
)
subgroup_id = db.Column(
UUIDType,
db.ForeignKey(Group.id, onupdate='CASCADE', ondelete='CASCADE'),
nullable=False
)
group = db.relationship(
Group, foreign_keys=[group_id], backref='subgroupsm2m')
subgroup = db.relationship(
Group, foreign_keys=[subgroup_id], backref='supergroupsm2m')
def __repr__(self):
"""String representation of the model."""
return f'<{self.group}: {self.subgroup}>'
[docs]class GroupRelationshipM2M(db.Model, Timestamp):
"""Many-to-many model for Group Relationships."""
__tablename__ = 'grouprelationshipm2m'
__table_args__ = (
PrimaryKeyConstraint('relationship_id', 'subrelationship_id',
name='pk_grouprelationshipm2m'),
)
relationship_id = db.Column(
UUIDType,
db.ForeignKey(
GroupRelationship.id, onupdate="CASCADE", ondelete="CASCADE"),
nullable=False
)
subrelationship_id = db.Column(
UUIDType,
db.ForeignKey(
GroupRelationship.id, onupdate="CASCADE", ondelete="CASCADE"),
nullable=False
)
relationship = db.relationship(
GroupRelationship,
foreign_keys=[relationship_id],
backref='subrelationshipsm2m')
subrelationship = db.relationship(
GroupRelationship,
foreign_keys=[subrelationship_id],
backref='superrelationshipsm2m')
def __repr__(self):
"""String representation of the model."""
return f'<{self.relationship}: {self.subrelationship}>'