Example model/controller and setup code; controller sets default values for key expiration and key id

01_models.py

import pendulum

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey

from bloop import UUID, Binary, Column
from bloop.ext.pendulum import DateTime


class PublicKeyType(Binary):
    """Stored in Dynamo in DER.  Locally, an RSAPublicKey"""
    python_type = RSAPublicKey

    def dynamo_load(self, value: str, *, context=None, **kwargs) -> RSAPublicKey:
        value = super().dynamo_load(value, context=context, **kwargs)
        return serialization.load_der_public_key(
            data=value,
            backend=default_backend())

    def dynamo_dump(self, value: RSAPublicKey, *, context=None, **kwargs) -> str:
        value = value.public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.SubjectPublicKeyInfo)
        return super().dynamo_dump(value, context=context, **kwargs)


class Key(BaseModel):
    class Meta:
        table_name = "keys"
    user_id = Column(UUID, hash_key=True, name='u')
    key_id = Column(UUID, range_key=True, name='k')
    public = Column(PublicKeyType, name='p')
    until = Column(DateTime, name='e')

    @property
    def is_expired(self):
        return pendulum.now() > self.until


KEY_NOT_EXIST = Key.user_id.is_(None) & Key.key_id.is_(None)

02_controllers.py

import bloop
import pendulum
from models import Key, KEY_NOT_EXIST


class KeyManager:
    def __init__(self, engine: bloop.Engine):
        self.engine = engine

    def create(self, user_id: uuid.UUID, public: bytes) -> Key:
        key = Key(
            user_id=user_id, public=public,
            # Set Defaults
            until=pendulum.now().add(hours=1),
            key_id=uuid.uuid4())
        self.engine.save(key, condition=KEY_NOT_EXIST)
        return key
    
    def get(self, user_id: uuid.UUID, key_id: uuid.UUID) -> Key:
        key = Key(user_id=user_id, key_id=key_id)
        self.engine.load(key)
        return key

03_entry_point.py

import boto3
import bloop
from models import Key
from controllers import KeyManager

session = boto3.session.Session(profile_name="PROFILE_NAME_FROM_CONFIG")
engine = bloop.Engine(
    dynamodb=session.client("dynamodb"),
    dynamodbstreams=session.client("dynamodbstreams"))
engine.bind(Key)

controller = KeyManager(engine)

# TODO hook up to some WSGI resource-based view
key_resource = {
    "create": controller.create,
    "get": controller.get
}