Skip to content

Instantly share code, notes, and snippets.

@numberoverzero
Last active January 27, 2019 15:13
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save numberoverzero/63017c1a32838cc454bc350425953581 to your computer and use it in GitHub Desktop.
Save numberoverzero/63017c1a32838cc454bc350425953581 to your computer and use it in GitHub Desktop.
Helpers to partition a single table as demonstrated in https://youtu.be/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, **kwargs):
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, **kwargs)
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment