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 MV_LINK_0(BaseMatViewsParity):
__tablename__ = 'mv_link_0'
#: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, index=True)
#: int: number of link face to list
link_key = Column(Integer)
[docs]class MV_LINK_1(BaseMatViewsParity):
__tablename__ = 'mv_link_1'
#: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, index=True)
#: int: number of link face to list
link_key = Column(Integer)
[docs]class MV_UNLINK_0(BaseMatViewsParity):
__tablename__ = 'mv_unlink_0'
#: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, index=True)
#: int: number of link face to list
unlink_key = Column(Integer)
[docs]class MV_UNLINK_1(BaseMatViewsParity):
__tablename__ = 'mv_unlink_1'
#: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, index=True)
#: int: number of link face to list
unlink_key = Column(Integer)
[docs]class MV_LINK(BaseMatViews):
__tablename__ = 'mv_link'
#: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, index=True)
#: int: number of link face to list
link_key = Column(Integer)
[docs]class MV_UNLINK(BaseMatViews):
__tablename__ = 'mv_unlink'
#: str: list id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
list_id = Column(String(36), primary_key=True, index=True)
#: int: number of link face to list
unlink_key = Column(Integer)
[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)
