PynamoDB same model with multipe databases

Question

The way PynamoDB is implemented is that it looks to a specific single DynamoDB table:

class UserModel(Model):
    class Meta:
        # Specific table.
        table_name = 'dynamodb-user'
        region = 'us-west-1'

The way my infrastructure works is that it has as many dynamodb tables as I have clients, so a single Lambda function has to deal with any amount of separate tables that are identical in structure e.g. represent "UserModel". I can't specify a concrete one.

How would I make this model definition dynamic?

Thanks!

Answer 1

Open-sourced a solution that is tested and works.

https://github.com/Biomapas/B.DynamoDbCommon/blob/master/b_dynamodb_common/models/model_type_factory.py

Read README.md for more details.

Code:

from typing import TypeVar, Generic, Type

from pynamodb.models import Model

T = TypeVar('T')


class ModelTypeFactory(Generic[T]):
    def __init__(self, model_type: Type[T]):
        self.__model_type = model_type

        # Ensure that given generic belongs to pynamodb.Model class.
        if not issubclass(model_type, Model):
            raise TypeError('Given model type must inherit from pynamodb.Model class!')

    def create(self, custom_table_name: str, custom_region: str) -> Type[T]:
        parent_class = self.__model_type

        class InnerModel(parent_class):
            class Meta:
                table_name = custom_table_name
                region = custom_region

        return InnerModel

Answer 2

Possible solution:

def create_user_model(table_name: str, region: str):
    return type("UserModel", (Model,), {
        "key" : UnicodeAttribute(hash_key=True),
        "range_key" : UnicodeAttribute(range_key=True),
        # Place for other keys
        "Meta": type("Meta", (object,), {
            "table_name": table_name,
            "region": region,
            "host": None,
            "billing_mode": 'PAY_PER_REQUEST',
        })
    })
UserModel_dev = create_user_model("user_model_dev", "us-west-1")
UserModel_prod = create_user_model("user_model_prod", "us-west-1")

Update:

A cleaner version:

class UserModel(Model):
    key = UnicodeAttribute(hash_key=True)
    range_key = UnicodeAttribute(range_key=True)

    @staticmethod
    def create(table_name: str, region: str):
        return type("UserModelDynamic", (UserModel,), {
            "Meta": type("Meta", (object,), {
                "table_name": table_name,
                "region": region,
                "host": None,
                "billing_mode": 'PAY_PER_REQUEST',
            })
        })