Source code for luna_faces.faces_db.models

from typing import Any

from sqlalchemy import Column as SaColumn, ForeignKey, Index, Integer, LargeBinary, Sequence, String, TIMESTAMP, or_
from sqlalchemy.dialects.oracle import RAW
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.elements import BinaryExpression
from sqlalchemy.sql.operators import ColumnOperators, mod

from crutches_on_wheels.utils.db_functions import currentDBTimestamp
from utils.enums import SampleType
from . import metadata
from .models_config import DBConfig

Base = declarative_base()
BaseMatViews = declarative_base()
BaseMatViewsParity = declarative_base()

if DBConfig.databaseNumber == 0:
    LINK_SEQUENCE = Sequence('link_key')
    UNLINK_SEQUENCE = Sequence('unlink_key')
else:
    LINK_SEQUENCE = Sequence('link_key', start=DBConfig.databaseNumber, increment=2)
    UNLINK_SEQUENCE = Sequence('unlink_key', start=DBConfig.databaseNumber, increment=2)

ALL_ATTRIBUTES_NAMES = ('descriptor', 'gender', 'age', 'ethnicity')

# database list max size, also database max size of list with list ids per link keys getting request
DB_LIST_LIMIT = 10 ** 3

# 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 Column(SaColumn): """ Release some pretty methods for sqlalchemy.Column. """
[docs] def in_(self, other: Any) -> or_: """ Split large ".in_" calls. Args: other: some values in list (or tuple) or select statement Returns: prepared *or* filter or just called *in_* """ if isinstance(other, set): other = list(other) if type(other) in (list, tuple) and DBConfig.dbType == 'oracle' and len(other) > DB_LIST_LIMIT: return or_(*(SaColumn.in_(self, other[index:index + DB_LIST_LIMIT]) for index in range(0, len(other), DB_LIST_LIMIT))) return SaColumn.in_(self, other)
def __matmul__(self, value: Any): """ Optional operator "eq". >>> accountId = None >>> Attribute.account_id @ accountId ... True >>> accountId = 'd58dcae8-3299-445f-b665-c877ea983aa4' >>> type(Attribute.account_id @ accountId) is BinaryExpression ... True Args: value: value for compare Returns: True if is None otherwise self == value """ return self == value if value is not None else True setattr(ColumnOperators, '__matmul__', __matmul__) #: hack for sqlalchemy
[docs]class Attribute(Base): """ Database table model for attributes. """ __tablename__ = 'attribute' Base.metadata = metadata #: str: face id, uuid. face_id = Column(String(36), ForeignKey('face.face_id', ondelete='CASCADE'), primary_key=True) #: int: gender. 0 - woman, 1- man gender = Column(Integer) #: int: how the gender was obtained gender_obtaining_method = Column(Integer) #: int: gender version gender_version = Column(Integer) #: int: age age = Column(Integer) #: int: how the gender was obtained age_obtaining_method = Column(Integer) #: int: age version age_version = Column(Integer) #: int: ethnicity, enum luna_faces.crutches_on_wheels.maps.vl_maps.ETHNIC_MAP ethnicity = Column(Integer) #: int: how the ethnicity was obtained ethnicity_obtaining_method = Column(Integer) #: int: ethnicity version ethnicity_version = Column(Integer) #: DateTime: date and time of creating attributes create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, index=True, nullable=False) #: int: descriptor samples generation descriptor_samples_generation = Column(Integer, nullable=False)
[docs]class Descriptor(Base): """ Database table model for descriptors. """ __tablename__ = 'descriptor' Base.metadata = metadata #: int: descriptor version descriptor_version = Column(Integer, primary_key=True) #: str: face id, uuid. face_id = Column(String(36), ForeignKey('attribute.face_id', ondelete='CASCADE'), primary_key=True, index=True) #: str: binary descriptor. Type depends on faces_db type if DBConfig.dbType == 'oracle': descriptor = Column(RAW(1024)) else: descriptor = Column(LargeBinary) #: int: how the descriptor was obtained descriptor_obtaining_method = Column(Integer) #: int: descriptor generation descriptor_generation = Column(Integer, nullable=False)
[docs]class Face(Base): """ Database table model for faces. """ __tablename__ = 'face' Base.metadata = metadata #: str: face id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" face_id = Column(String(36), primary_key=True) #: str: account uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", to which this face belong account_id = Column(String(36)) # : str: event id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", reference to event which created # the face event_id = Column(String(36), index=True) #: str: client info about the face user_data = Column(String(128), index=True) #: DateTime: date and time of creating face create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, index=True) #: DateTime: date and time of last changed of the face last_update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP) #: str: external id of the face, if it has its own mapping in external system external_id = Column(String(36), index=True) #: str: url to image, that represents the face avatar = Column(String(256), comment="str: the face avatar image url")
[docs]class List(Base): """ Database table model for lists. Warnings: `trg_lists_deletion_log` ~--- after delete trigger for logging """ __tablename__ = 'list' Base.metadata = metadata #: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" list_id = Column(String(36), primary_key=True) #: str: account uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", to which this list belong account_id = Column(String(36)) #: str: client info about the list user_data = Column(String(128), index=True) #: DateTime: date and time of creating list create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, index=True) #: DateTime: date and time of last changed of the list last_update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP)
[docs]class ListFace(Base): """ Database table model for links between faces and lists. """ __tablename__ = 'list_face' Base.metadata = metadata #: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" list_id = Column(String(36), ForeignKey('list.list_id', ondelete='CASCADE'), primary_key=True) #: str: face id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" face_id = Column(String(36), ForeignKey('face.face_id', ondelete='CASCADE'), primary_key=True, index=True) #: DateTime: date and time of last attach/detach face to list last_update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP) #: int: number of link face to list link_key = Column(Integer, LINK_SEQUENCE, nullable=False) if DBConfig.databaseNumber > 0: #: index for parity link keys link_key_index = Index('link_key_func_index', list_id, mod(link_key, 2), link_key) else: #: index link keys link_key_index = Index('list_id_link_key_index', list_id, link_key) #: index for load delta into matcher matcher_load_delta_index = Index('matcher_load_delta_index', link_key)
[docs]class UnlinkAttributesLog(Base): """ Database table model for history attach and detach attributes to lists. """ __tablename__ = 'unlink_attributes_log' Base.metadata = metadata #: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" list_id = Column(String(36), ForeignKey('list.list_id', ondelete='CASCADE')) #: str: face id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" face_id = Column(String(36), index=True) #: int: number of link face to list link_key = Column(Integer, index=True, unique=True) #: int: number of unlink face to list unlink_key = Column(Integer, UNLINK_SEQUENCE, primary_key=True) #: DateTime: date and time of detach attributes from list update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP) if DBConfig.databaseNumber > 0: #: index for parity unlink keys unlink_key_index = Index('unlink_key_func_index', list_id, mod(unlink_key, 2), unlink_key) else: #: index unlink keys unlink_key_index = Index('list_id_unlink_key_index', list_id, unlink_key)
[docs]class Sample(Base): """ Database model to store samples. """ __tablename__ = 'sample' Base.metadata = metadata #: str: parent face id, uuid. face_id = Column(String(36), ForeignKey(Attribute.face_id, ondelete='CASCADE'), primary_key=True, comment='uuid: face id') #: int: enum for sample type, check 'SampleType' in luna_faces/utils/enums.py type = Column(Integer, primary_key=True, comment=f'int: sample type: ' f'{", ".join(f"{k} - {v.value}" for k, v in SampleType.__members__.items())}') #: str: id(uuid) of warp sample_id = Column(String(36), comment='uuid: sample id', primary_key=True)
[docs]class ListsDeletionLog(Base): """ Database table model for lists deletions history After delete trigger trg_lists_deletion_log` (table `list`) insert a data. """ __tablename__ = 'lists_deletion_log' Base.metadata = metadata #: str: deletion id deletion_id = Column(Integer, Sequence('list_deletion_row_id'), primary_key=True, nullable=False, comment='int: record id', autoincrement=False) #: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" list_id = Column(String(36), index=True, nullable=False, comment='uuid: id of deleted list') #: str: account uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", to which this face belong account_id = Column(String(36), nullable=False, comment='uuid: list account id') #: DateTime: date list removig deletion_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, index=True, nullable=False, comment="date: list removal time") #: DateTime: create time list create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment='date: list create time', nullable=False)