from datetime import datetime
from luna_db_tools.functions import currentDBTimestamp
from sqlalchemy import Column as SaColumn, ForeignKey, Index, Integer, LargeBinary, Sequence, String, TIMESTAMP, Text, 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 sqlalchemy.dialects.oracle import CLOB
from sqlalchemy.dialects.postgresql import TEXT
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.cow.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 RequestsCache(Base):
"""
Model that provides cached data for various often used requests
"""
__tablename__ = "requests_cache"
# timestamp: cache creation time
created_at = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, nullable=True)
# str: unique name of the cache
name = Column(String(128), nullable=False, primary_key=True)
# str: Encrypted value of cached data
if DBConfig.dbType == "oracle":
value = Column(CLOB, nullable=True)
else:
value = Column(TEXT, nullable=True)
[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)
#: str: descriptor encryption hash. Type depends on faces_db type
if DBConfig.dbType == "oracle":
encryption_hash = Column(RAW(32), nullable=True)
else:
encryption_hash = Column(LargeBinary, nullable=True)
#: 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
)