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

168 lines
5.3 KiB
Python
Raw Normal View History

from itertools import chain
from operator import attrgetter
from uuid import uuid4
from citext import CIText
2022-04-06 10:35:08 +00:00
from sqlalchemy import Column
from sqlalchemy import Enum as DBEnum
from sqlalchemy import ForeignKey, Unicode, UniqueConstraint
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.declarative import declared_attr
2018-09-20 07:28:52 +00:00
from sqlalchemy.orm import backref, relationship, validates
from sqlalchemy_utils import EmailType, PhoneNumberType
2019-01-23 15:55:04 +00:00
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.inventory import Inventory
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing
2018-09-07 10:38:02 +00:00
from ereuse_devicehub.resources.user.models import User
2023-03-21 11:08:13 +00:00
from ereuse_devicehub.teal import enums
from ereuse_devicehub.teal.db import (
INHERIT_COND,
POLYMORPHIC_ID,
POLYMORPHIC_ON,
check_lower,
)
from ereuse_devicehub.teal.marshmallow import ValidationError
2018-09-07 10:38:02 +00:00
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'))
2022-04-06 11:49:15 +00:00
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.'),
2022-04-06 10:35:08 +00:00
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
2022-04-06 10:35:08 +00:00
return sorted(
chain(self.actions_agent, self.actions_to), key=attrgetter('created')
)
2018-09-20 07:28:52 +00:00
@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):
2022-04-06 10:35:08 +00:00
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,
)
2019-01-23 15:55:04 +00:00
def __init__(self, name: str, **kwargs) -> None:
super().__init__(**kwargs, name=name)
@classmethod
def get_default_org_id(cls) -> UUID:
"""Retrieves the default organization."""
2019-01-23 15:55:04 +00:00
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))
2022-04-06 10:35:08 +00:00
active_org = relationship(
Organization, primaryjoin=active_org_id == Organization.id
)
user_id = Column(UUID(as_uuid=True), ForeignKey(User.id), unique=True)
2022-04-06 10:35:08 +00:00
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.
"""
2022-04-06 10:35:08 +00:00
id = Column(Unicode(), check_lower('id'))
2022-04-06 10:35:08 +00:00
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.
"""
2022-04-06 10:35:08 +00:00
pass
class System(Individual):
pass