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 sqlalchemy.orm import validates
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)
#: int: lambda workers
workers = Column(INTEGER, server_default=text("1"))
#: smallint: whether gpu is enabled
gpu = Column(SMALLINT, server_default=text("0"))
#: int: lambda pod count
pod_count = Column(INTEGER, server_default=text("1"))
#: int: cpu units limit
cpu_limit = Column(INTEGER, server_default=text("1000"))
#: int: ram limit
ram_limit = Column(INTEGER, server_default=text("1"))
#: int: cpu unit request
cpu_request = Column(INTEGER, server_default=text("500"), nullable=True)
#: int: ram request
ram_request = Column(INTEGER, server_default=text("1"), nullable=True)
#: str: namespace
namespace = Column(String(63), server_default=text("'default'"), nullable=False)
#: dict: lambda selector
selector = Column(CLOB if DBConfig.dbType == "oracle" else String, server_default=text("'{}'"))
@validates('workers')
def validate_workers(self, key, value):
if not 1 < value < 32:
raise ValueError(f'Invalid workers {value}, workers must be integer between 1 and 32.')
return value