dynamodb.py

import functools
import inspect

from nrlf.core.boto import get_dynamodb_resource
from nrlf.core.config import Config
from nrlf.core.dynamodb.repository import DocumentPointerRepository
from nrlf.core.types import DynamoDBServiceResource


def create_document_pointer_table(config: Config, dynamodb: DynamoDBServiceResource):
    return dynamodb.create_table(
        TableName=config.TABLE_NAME,
        KeySchema=[
            {"AttributeName": "pk", "KeyType": "HASH"},
            {"AttributeName": "sk", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "pk", "AttributeType": "S"},
            {"AttributeName": "sk", "AttributeType": "S"},
            {"AttributeName": "patient_key", "AttributeType": "S"},
            {"AttributeName": "patient_sort", "AttributeType": "S"},
            {"AttributeName": "masterid_key", "AttributeType": "S"},
        ],
        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
        GlobalSecondaryIndexes=[
            {
                "IndexName": "patient_gsi",
                "KeySchema": [
                    {"AttributeName": "patient_key", "KeyType": "HASH"},
                    {"AttributeName": "patient_sort", "KeyType": "RANGE"},
                ],
                "Projection": {"ProjectionType": "ALL"},
            },
            {
                "IndexName": "masterid_gsi",
                "KeySchema": [
                    {"AttributeName": "masterid_key", "KeyType": "HASH"},
                ],
                "Projection": {"ProjectionType": "ALL"},
            },
        ],
    )


def mock_repository(func):
    @functools.wraps(func)
    def wrapped_function(*args, **kwargs):
        config = Config()
        dynamodb = get_dynamodb_resource()
        create_document_pointer_table(config, dynamodb)

        repository = DocumentPointerRepository(table_name=config.TABLE_NAME)

        return func(*args, **kwargs, repository=repository)

    # Remove 'repository' from the visible signature so pytest doesn't try
    # to inject it as a test parameter (it's already injected by this decorator).
    sig = inspect.signature(func)
    params_minus_repository = [
        p for name, p in sig.parameters.items() if name != "repository"
    ]
    wrapped_function.__signature__ = sig.replace(parameters=params_minus_repository)

    return wrapped_function