Source code for luna_lambda.db.lambda_db_tools.models.lambda_models

from typing import Any

from sqlalchemy import INTEGER, SMALLINT, TIMESTAMP, Column as SaColumn, String, text
from sqlalchemy.dialects.oracle import CLOB
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.operators import ColumnOperators

from .config import DBConfig

DB_CURRENT_TIMESTAMP = text(
    "SYS_EXTRACT_UTC(SYSTIMESTAMP)"
    if DBConfig.dbType == "oracle"
    else "(now() at time zone 'utc')"
)

Base = declarative_base()


[docs] class Column(SaColumn): """Release some pretty methods for sqlalchemy.Column.""" 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 Lambda(Base): """ Lambda model for database. """ __tablename__ = "lambda" #: str: lambda id, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" id = Column(String(36), primary_key=True) #: str: lambda name, uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" name = Column(String(36), index=True, unique=True) #: str(256): description description = Column(String(256)) #: int: lambda version version = Column(INTEGER, server_default=text("1")) #: str: account uuid in format "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", to which this lambda belong account_id = Column(String(36)) #: DateTime: date and time of creating lambda create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP) #: DateTime: date and time of last changed of the lambda last_update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP) #: smallint: status status = Column(SMALLINT, server_default=text("0")) #: enum: lambda type lambda_type = Column( String(10), server_default=text("'standalone'"), nullable=False ) #: str: list with additional user docker commands user_commands = Column(CLOB if DBConfig.dbType == "oracle" else String, nullable=True) #: str: base lambda image name base_image_name = Column(String(128), server_default=text("'lpa-lambda-base'"), nullable=False) #: str: base lambda image tag base_image_tag = Column(String(36), server_default=text("'unknown'"), nullable=False)