from crutches_on_wheels.utils.db_functions import currentDBTimestamp
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 typing import Any
from .config import DBConfig
from .enums import SampleType
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)
# 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
Base = declarative_base()
BaseMatViews = declarative_base()
BaseMatViewsParity = declarative_base()
[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"
#: 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"
#: 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"
#: 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"
#: 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"
#: 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"
#: 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"
#: 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"
#: 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
)
