"""
Backport3 database models.
"""
import typing
from luna_db_tools.functions import currentDBTimestamp
from sqlalchemy import Column, String, ForeignKey, Boolean, TIMESTAMP, Integer, Index
from sqlalchemy import MetaData
from sqlalchemy.dialects.oracle import CLOB
from sqlalchemy.dialects.postgresql import TEXT
from sqlalchemy.ext.declarative import declarative_base
from .config import DBConfig
Base = declarative_base()
metadata = MetaData()
# raw sql function to get utc timestamp for creating/updating table columns
DB_CURRENT_TIMESTAMP = currentDBTimestamp(DBConfig.dbType) if DBConfig.dbType else None
[docs]
class Account(Base):
"""
Database table model for account.
"""
__tablename__ = 'account'
#: uuid4: account id.
account_id = Column(String(36), primary_key=True)
#: bool: account status (blocked or not).
active = Column(Boolean)
[docs]
class AccountToken(Base):
"""
Token table.
"""
__tablename__ = 'account_token'
#: uuid4: token id
token_id = Column(String(36), primary_key=True)
#: uuid4: account id, token is linked to
account_id = Column(String(36), ForeignKey('account.account_id', ondelete='CASCADE'))
#: str: token
token = Column(CLOB() if DBConfig.dbType == "oracle" else String())
[docs]
class Person(Base):
"""
Database table model for persons.
"""
__tablename__ = 'person'
__table_args__ = {'comment': 'Table of persons.'}
#: str: person id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
person_id = Column(String(36), primary_key=True, comment="uuid4: person id")
#: str: descriptor id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
account_id = Column(String(36), ForeignKey('account.account_id', ondelete='CASCADE'), index=True,
comment="uuid4: id of the account to which the person belongs")
#: str: client info about the face
user_data = Column(String(128), index=True, comment="str: person user data")
#: DateTime: date and time of creating person
create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment="date: person create time",
nullable=False)
#: str: external id of the person, if it has its own mapping in external system
external_id = Column(String(36), index=True, comment="str: person external id")
[docs]
class PersonsList(Base):
"""
Database table model for person lists.
"""
__tablename__ = 'persons_list'
__table_args__ = {'comment': 'Table of person lists.'}
#: str: list id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, comment="uuid4: list id")
#: str: account id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
account_id = Column(String(36), ForeignKey('account.account_id', ondelete='CASCADE'), index=True,
comment="uuid4: id of the account to which the list belongs")
#: DateTime: date and time of creating list
create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment="date: list create time",
nullable=False)
#: index for pagination by list
create_persons_list_time_index = Index('create_persons_list_time_index', account_id, create_time)
[docs]
class DescriptorsList(Base):
"""
Database table model for descriptors lists.
"""
__tablename__ = 'descriptors_list'
__table_args__ = {'comment': 'Table of descriptors lists.'}
#: str: list id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, comment="uuid4: list id")
#: str: account id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
account_id = Column(String(36), ForeignKey('account.account_id', ondelete='CASCADE'), index=True,
comment="uuid4: id of the account to which the list belongs")
#: DateTime: date and time of creating list
create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment="list: person create time",
nullable=False)
#: index for pagination by list
create_descr_list_time_index = Index('create_descr_list_time_index', account_id, create_time)
[docs]
class ListPerson(Base):
"""
Database table model for links between persons and lists.
"""
__tablename__ = 'list_person'
__table_args__ = {'comment': 'Relationship table for lists and persons.'}
#: str: list id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
list_id = Column(String(36), ForeignKey('persons_list.list_id', ondelete='CASCADE'), primary_key=True,
comment="uuid4: list id")
#: str: person id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
person_id = Column(String(36), ForeignKey('person.person_id', ondelete='CASCADE',
name="list_person_list_id_fkey"), primary_key=True, index=True,
comment="uuid4: person id")
[docs]
class PersonFace(Base):
"""
Database table model for links between persons and faces.
"""
__tablename__ = 'person_face'
__table_args__ = {'comment': 'Relationship table for persons and faces.'}
#: str: person id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
person_id = Column(String(36), ForeignKey('person.person_id', ondelete='CASCADE'), primary_key=True,
comment="uuid4: person id")
#: str: face id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
face_id = Column(String(36), primary_key=True, comment="uuid4: face id", unique=True)
[docs]
class Handler(Base):
"""
Database table model for handlers.
"""
__tablename__ = "handler"
#: str: handler id, uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx"
handler_id = Column(String(36), nullable=False)
#: str: account uuid4 in format "xxxxxxxx-xxxx-4xxx-{8-9}xx-xxxxxxxxxxxx", to which this handler belong
account_id = Column(String(36), ForeignKey('account.account_id', ondelete='CASCADE'), primary_key=True)
#: DateTime: date and time of creating handler
create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment="date: handler create time",
nullable=False)
#: DateTime: date and time of creating face
last_update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment="date: handler last update time",
nullable=False, index=True)
#: str: json with policies
policies = Column(CLOB if DBConfig.dbType == "oracle" else TEXT, nullable=False)
#: integer: handler type (extractor - 0, identify - 1, verify - 2)
type = Column(Integer, primary_key=True)
#: int: handler version
version = Column(Integer, nullable=False)
[docs]
@classmethod
def getColumnNames(cls) -> typing.List[str]:
"""
Get all column name of table.
Returns:
list of column name in order as in db
"""
return cls.__table__.columns._data._list