Module kvalchemy.models

Home to models for KVAlchemy.

"""
Home to models for KVAlchemy.
"""
from datetime import datetime

from sqlalchemy import Column, ColumnElement, PrimaryKeyConstraint, event, or_
from sqlalchemy.dialects.mysql import LONGBLOB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.types import LargeBinary, PickleType, Unicode

from kvalchemy.time import db_now

KEY_MAX_LENGTH = 256
TAG_MAX_LENGTH = 256


class Base(DeclarativeBase):
    """
    The base class for all models.
    """

    pass


class ValueMixIn:
    """
    A mixin used to correspond with an object with a value attribute
    """

    value = Column(
        "value", PickleType(impl=LargeBinary().with_variant(LONGBLOB, "mysql"))
    )

    def __init__(self, value):
        self.value = value


class KVStore(Base, ValueMixIn):
    """
    The table for storing key-value pairs.
    """

    __tablename__ = "kvstore"

    __table_args__ = (PrimaryKeyConstraint("key", "tag", name="key_tag_unique"),)

    key: Mapped[str] = Column(Unicode(KEY_MAX_LENGTH))
    tag: Mapped[str] = Column(Unicode(TAG_MAX_LENGTH))

    # Naive datetime (though expected to be UTC)
    expire: Mapped[datetime] = mapped_column(nullable=True)

    @classmethod
    def non_expired_filter(cls) -> ColumnElement[bool]:
        """
        A filter that can be used to find all non expired key-value pairs.
        """
        return or_(KVStore.expire == None, KVStore.expire > db_now())

Classes

class Base (**kwargs: Any)

The base class for all models.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance's class are allowed. These could be, for example, any mapped columns or relationships.

Expand source code

class Base(DeclarativeBase):
    """
    The base class for all models.
    """

    pass

Ancestors

• sqlalchemy.orm.decl_api.DeclarativeBase
• sqlalchemy.inspection.Inspectable
• typing.Generic

Subclasses

• KVStore

Class variables

var metadata

var registry

class KVStore (**kwargs)

The table for storing key-value pairs.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance's class are allowed. These could be, for example, any mapped columns or relationships.

Expand source code

class KVStore(Base, ValueMixIn):
    """
    The table for storing key-value pairs.
    """

    __tablename__ = "kvstore"

    __table_args__ = (PrimaryKeyConstraint("key", "tag", name="key_tag_unique"),)

    key: Mapped[str] = Column(Unicode(KEY_MAX_LENGTH))
    tag: Mapped[str] = Column(Unicode(TAG_MAX_LENGTH))

    # Naive datetime (though expected to be UTC)
    expire: Mapped[datetime] = mapped_column(nullable=True)

    @classmethod
    def non_expired_filter(cls) -> ColumnElement[bool]:
        """
        A filter that can be used to find all non expired key-value pairs.
        """
        return or_(KVStore.expire == None, KVStore.expire > db_now())

Ancestors

• Base
• sqlalchemy.orm.decl_api.DeclarativeBase
• sqlalchemy.inspection.Inspectable
• typing.Generic
• ValueMixIn

Static methods

def non_expired_filter() ‑> sqlalchemy.sql.elements.ColumnElement[bool]

A filter that can be used to find all non expired key-value pairs.

Expand source code

@classmethod
def non_expired_filter(cls) -> ColumnElement[bool]:
    """
    A filter that can be used to find all non expired key-value pairs.
    """
    return or_(KVStore.expire == None, KVStore.expire > db_now())

Instance variables

var expire : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]

Expand source code

def __get__(
    self, instance: Optional[object], owner: Any
) -> Union[InstrumentedAttribute[_T_co], _T_co]:
    if instance is None:
        return self

    dict_ = instance_dict(instance)
    if self.impl.supports_population and self.key in dict_:
        return dict_[self.key]  # type: ignore[no-any-return]
    else:
        try:
            state = instance_state(instance)
        except AttributeError as err:
            raise orm_exc.UnmappedInstanceError(instance) from err
        return self.impl.get(state, dict_)  # type: ignore[no-any-return]

var key : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]

Expand source code

def __get__(
    self, instance: Optional[object], owner: Any
) -> Union[InstrumentedAttribute[_T_co], _T_co]:
    if instance is None:
        return self

    dict_ = instance_dict(instance)
    if self.impl.supports_population and self.key in dict_:
        return dict_[self.key]  # type: ignore[no-any-return]
    else:
        try:
            state = instance_state(instance)
        except AttributeError as err:
            raise orm_exc.UnmappedInstanceError(instance) from err
        return self.impl.get(state, dict_)  # type: ignore[no-any-return]

var tag : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]

Expand source code

def __get__(
    self, instance: Optional[object], owner: Any
) -> Union[InstrumentedAttribute[_T_co], _T_co]:
    if instance is None:
        return self

    dict_ = instance_dict(instance)
    if self.impl.supports_population and self.key in dict_:
        return dict_[self.key]  # type: ignore[no-any-return]
    else:
        try:
            state = instance_state(instance)
        except AttributeError as err:
            raise orm_exc.UnmappedInstanceError(instance) from err
        return self.impl.get(state, dict_)  # type: ignore[no-any-return]

var value : Union[sqlalchemy.orm.attributes.InstrumentedAttribute[+_T_co], +_T_co]

Expand source code

def __get__(
    self, instance: Optional[object], owner: Any
) -> Union[InstrumentedAttribute[_T_co], _T_co]:
    if instance is None:
        return self

    dict_ = instance_dict(instance)
    if self.impl.supports_population and self.key in dict_:
        return dict_[self.key]  # type: ignore[no-any-return]
    else:
        try:
            state = instance_state(instance)
        except AttributeError as err:
            raise orm_exc.UnmappedInstanceError(instance) from err
        return self.impl.get(state, dict_)  # type: ignore[no-any-return]

class ValueMixIn (value)

A mixin used to correspond with an object with a value attribute

Expand source code

class ValueMixIn:
    """
    A mixin used to correspond with an object with a value attribute
    """

    value = Column(
        "value", PickleType(impl=LargeBinary().with_variant(LONGBLOB, "mysql"))
    )

    def __init__(self, value):
        self.value = value

Subclasses

• KVStore

Class variables

var value