Skip to content

Late bind to host, region and other defaultable settings #1275

@sirianni

Description

@sirianni

I am trying to use PynamoDB in unit tests with testcontainers and the dynamodb-local container.

In this setup, an ephemeral port is allocated for the dynamodb-local container. The ephemeral port is only available after the container starts. I use this ephemeral port to set the host URL by monkey-patching pyndamodb.settings.

This works well. My problem, however, is that the PynamoDB MetaModel class copies in the default host from its settings when the Model class is declared. This means I need to define the class WidgetModel in my test case after the test container fixture is started. While this is OK in this toy example, my actual Model classes are defined in application code and imported before the test container is started - therefore they get the incorrect host setting.

My preference would be for the settings inside the Meta class (host, region, etc.) to be left as None if not set and then resolved from the pynamodb.settings object lazily at the time the connection is created.

This pytest code should be runnable locally with the following libraries

dependencies = [
    "boto3==1.38.16",
    "pynamodb>=6.1.0",
    "testcontainers[k3s]>=4.10.0",
]
import logging
import os
from types import SimpleNamespace
from typing import cast

import pytest
from botocore.exceptions import BotoCoreError
from pynamodb import settings
from pynamodb.attributes import UnicodeAttribute
from pynamodb.connection import Connection
from pynamodb.models import Model
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_container_is_ready
from types_boto3_dynamodb.client import DynamoDBClient

logger = logging.getLogger(__name__)


class DynamoDbContainer(DockerContainer):
    """
    DynamoDB local container for testing.
    """

    def __init__(
        self, image: str = "amazon/dynamodb-local", port: int = 8000, **kwargs
    ) -> None:
        super().__init__(image, **kwargs)
        self.port = port
        self.with_exposed_ports(self.port)

    def start(self) -> "DynamoDbContainer":
        super().start()

        self._patch_pynamodb_settings()
        self._connect()

        return self

    @wait_container_is_ready(ConnectionError, BotoCoreError)
    def _connect(self) -> None:
        try:
            conn = self.connect()
            boto_client: DynamoDBClient = cast(DynamoDBClient, conn.client)
            boto_client.describe_limits()
            logger.info("Connection to DynamoDB local container succeeded")
        except Exception as e:
            logger.error(f"Failed to connect to DynamoDB local container: {e}")
            raise

    def get_connection_url(self) -> str:
        host = self.get_container_host_ip()
        port = self.get_exposed_port(self.port)
        return f"http://{host}:{port}"

    def connect(self) -> Connection:
        return Connection(
            host=self.get_connection_url(),
        )

    def _patch_pynamodb_settings(self) -> None:
        # https://github.com/pynamodb/PynamoDB/issues/1272
        os.environ["AWS_ACCESS_KEY_ID"] = "bogus"
        os.environ["AWS_SECRET_ACCESS_KEY"] = "bogus"

        override = SimpleNamespace(
            host=self.get_connection_url(),
            region="bogus",
        )
        logger.info(f"Patching PynamoDB settings with: {override}")

        # Monkey-patch PynamoDB override_settings with our config
        # See https://github.com/pynamodb/PynamoDB/issues/555
        settings.override_settings = override


@pytest.fixture(scope="session")
def dynamodb():
    with DynamoDbContainer() as container:
        yield container


def test_crud(dynamodb):
    # Must be created after the container starts so that `host` and other attributes
    # of the metaclass are set from our overridden PynamoDB settings.
    class WidgetModel(Model):
        class Meta:  # pyright: ignore[reportIncompatibleVariableOverride]
            table_name = "widgets"
            read_capacity_units = 1
            write_capacity_units = 1

        id = UnicodeAttribute(hash_key=True)
        name = UnicodeAttribute()

    WidgetModel.create_table(wait=True)

    widget = WidgetModel(id="id", name="Test Widget")
    widget.save()
    assert WidgetModel.get("id").name == "Test Widget"

    widget.name = "Updated Widget"
    widget.save()
    assert WidgetModel.get("id").name == "Updated Widget"

    widget.delete()
    assert WidgetModel.count("id") == 0

    WidgetModel.delete_table(wait=True)

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions