This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
devicehub-teal/ereuse_devicehub/resources/agent/models.py

142 lines
5.4 KiB
Python

from itertools import chain
from operator import attrgetter
from uuid import uuid4
from citext import CIText
from sqlalchemy import Column, Enum as DBEnum, ForeignKey, Unicode, UniqueConstraint
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import backref, relationship, validates
from sqlalchemy_utils import EmailType, PhoneNumberType
from teal import enums
from teal.db import INHERIT_COND, POLYMORPHIC_ID, POLYMORPHIC_ON, check_lower
from teal.marshmallow import ValidationError
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.inventory import Inventory
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing
from ereuse_devicehub.resources.user.models import User
class JoinedTableMixin:
# noinspection PyMethodParameters
@declared_attr
def id(cls):
return Column(UUID(as_uuid=True), ForeignKey(Agent.id), primary_key=True)
class Agent(Thing):
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
type = Column(Unicode, nullable=False)
name = Column(CIText())
name.comment = """The name of the organization or person."""
tax_id = Column(Unicode(length=STR_SM_SIZE), check_lower('tax_id'))
tax_id.comment = """The Tax / Fiscal ID of the organization,
e.g. the TIN in the US or the CIF/NIF in Spain.
"""
country = Column(DBEnum(enums.Country))
country.comment = """Country issuing the tax_id number."""
telephone = Column(PhoneNumberType())
email = Column(EmailType, unique=True)
__table_args__ = (
UniqueConstraint(tax_id, country, name='Registration Number per country.'),
UniqueConstraint(tax_id, name, name='One tax ID with one name.'),
db.Index('agent_type', type, postgresql_using='hash')
)
@declared_attr
def __mapper_args__(cls):
"""Defines inheritance.
From `the guide <http://docs.sqlalchemy.org/en/latest/orm/
extensions/declarative/api.html
#sqlalchemy.ext.declarative.declared_attr>`_
"""
args = {POLYMORPHIC_ID: cls.t}
if cls.t == 'Agent':
args[POLYMORPHIC_ON] = cls.type
if JoinedTableMixin in cls.mro():
args[INHERIT_COND] = cls.id == Agent.id
return args
@property
def actions(self) -> list:
# todo test
return sorted(chain(self.actions_agent, self.actions_to), key=attrgetter('created'))
@validates('name')
def does_not_contain_slash(self, _, value: str):
if '/' in value:
raise ValidationError('Name cannot contain slash \'')
return value
def __repr__(self) -> str:
return '<{0.t} {0.name}>'.format(self)
class Organization(JoinedTableMixin, Agent):
default_of = db.relationship(Inventory,
uselist=False,
lazy=True,
backref=backref('org', lazy=True),
# We need to use this as we cannot do Inventory.foreign -> Org
# as foreign keys can only reference to one table
# and we have multiple organization table (one per schema)
foreign_keys=[Inventory.org_id],
primaryjoin=lambda: Organization.id == Inventory.org_id)
def __init__(self, name: str, **kwargs) -> None:
super().__init__(**kwargs, name=name)
@classmethod
def get_default_org_id(cls) -> UUID:
"""Retrieves the default organization."""
return cls.query.filter_by(default_of=Inventory.current).one().id
class Individual(JoinedTableMixin, Agent):
active_org_id = Column(UUID(as_uuid=True), ForeignKey(Organization.id))
active_org = relationship(Organization, primaryjoin=active_org_id == Organization.id)
user_id = Column(UUID(as_uuid=True), ForeignKey(User.id), unique=True)
user = relationship(User,
backref=backref('individuals', lazy=True, collection_class=set),
primaryjoin=user_id == User.id)
class Membership(Thing):
"""Organizations that are related to the Individual.
For example, because the individual works in or because is a member of.
"""
id = Column(Unicode(), check_lower('id'))
organization_id = Column(UUID(as_uuid=True), ForeignKey(Organization.id), primary_key=True)
organization = relationship(Organization,
backref=backref('members', collection_class=set, lazy=True),
primaryjoin=organization_id == Organization.id)
individual_id = Column(UUID(as_uuid=True), ForeignKey(Individual.id), primary_key=True)
individual = relationship(Individual,
backref=backref('member_of', collection_class=set, lazy=True),
primaryjoin=individual_id == Individual.id)
def __init__(self, organization: Organization, individual: Individual, id: str = None) -> None:
super().__init__(organization=organization,
individual=individual,
id=id)
__table_args__ = (
UniqueConstraint(id, organization_id, name='One member id per organization.'),
)
class Person(Individual):
"""A person in the system. There can be several persons pointing to
a real.
"""
pass
class System(Individual):
pass