bloop_shared_table_helper.py

# fragments from evaluating some toolkit classes to build shared models as used
# in DAT401: Advanced Design Patterns for DynamoDB
# https://www.youtube.com/watch?v=HaEPXoXVf2k

import functools
from typing import Type
from bloop.conditions import Condition, iter_columns
from bloop.models import BaseModel, Column, IMeta, bind_column


def default_hk_query(cls, engine, key=None, filter=None, projection="all", consistent=False, forward=True):
    key = Condition() | key
    hk = cls.Meta.hash_key
    if hk not in set(iter_columns(key)):
        key &= hk == hk.default()
    return engine.query(
        cls, key=key,
        filter=filter,
        projection=projection,
        consistent=consistent,
        forward=forward,
    )


def multimodel(base: Type[BaseModel], type_name = "PK_NAME") -> Type[BaseModel]:
    """
    .. code-block:: python

        >>> class AbstractBase(BaseModel):
        ...     class Meta:
        ...         table_name = "my_application"
        ...     _hk = Column(String, hash_key=True, dynamo_name="hk")
        ...     _rk = Column(String, range_key=True, dynamo_name="rk")
        ...
        >>> Base = multimodel(AbstractBase, type_name="HASH_KEY")
        >>> class User(Base):
        ...     HASH_KEY = "users"
        ...     name = Column(String, range_key=True)
        ...
        >>> class Order(Base):
        ...     HASH_KEY = "orders"
        ...     id = Column(String, range_key=True)
        ...
        >>> engine.bind(Base)
        
        >>> jane = User(name="Jane")
        >>> new_books = Order(id="jane--2018-02-14--X712540001")
        >>> engine.save(jane, new_books)
        >>> jane
        User(_hk="users", name="Jane")
        >>> book_order = Order.Meta.q(
        ...     engine, Order.id.begins_with("jane--")
        ... ).first()
        >>> book_order
        Order(_hk='orders', id='jane--2018-02-14--X712540001')
    """
    class _SharedBase(base):
        class Meta(IMeta):
            table_name = base.Meta.table_name

        def __init_subclass__(cls: Type["_SharedBase"], **kwargs):
            super().__init_subclass__(**kwargs)

            cls.Meta.table_name = base.Meta.table_name
            cls.Meta.q = functools.partial(default_hk_query, cls)

            hk = cls.Meta.hash_key
            hk._dynamo_name = base.Meta.hash_key.dynamo_name
            bind_column(cls, hk.name, hk, force=True)

            default_value = getattr(cls, type_name)
            hk.default = lambda: default_value

            rk = cls.Meta.range_key
            rk._dynamo_name = base.Meta.range_key.dynamo_name
            bind_column(cls, rk.name, rk, force=True)
            
    return _SharedBase