Skip to content

DynamoDB Repository

Repository

Manages all DynamoDB interactions for a single table.

Source code in templates/repository.py
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Repository:
    """Manages all DynamoDB interactions for a single table."""

    def __init__(self, table_name: str) -> None:
        """Initialize the repository with a DynamoDB table.

        Args:
            table_name: The name of the DynamoDB table to operate on.
        """
        self._table = resource("dynamodb").Table(table_name)

    def get_item(self, item_id: str) -> dict | None:
        """Retrieve an item by its ID.

        Args:
            item_id: The unique identifier of the item to retrieve.

        Returns:
            The item as a dictionary, or `None` if not found.
        """
        return self._table.get_item(Key={"id": item_id}).get("Item")

    def list_items(self) -> list[dict]:
        """Retrieve all items from the table.

        Returns:
            A list of all items in the table.
        """
        return self._table.scan().get("Items", [])

    def put_item(self, item: dict) -> None:
        """Write an item to the table, replacing any existing item with the same key.

        Args:
            item: A dictionary representing the item to store.
        """
        self._table.put_item(Item=item)

    def delete_item(self, keys: object) -> None:
        """Delete an item from the table by its primary key.

        Args:
            keys: A mapping of key attribute names to their values.
        """
        self._table.delete_item(Key=keys)

__init__(table_name)

Initialize the repository with a DynamoDB table.

Parameters:

Name Type Description Default
table_name str

The name of the DynamoDB table to operate on.

required
Source code in templates/repository.py
 7
 8
 9
10
11
12
13
def __init__(self, table_name: str) -> None:
    """Initialize the repository with a DynamoDB table.

    Args:
        table_name: The name of the DynamoDB table to operate on.
    """
    self._table = resource("dynamodb").Table(table_name)

delete_item(keys)

Delete an item from the table by its primary key.

Parameters:

Name Type Description Default
keys object

A mapping of key attribute names to their values.

required
Source code in templates/repository.py
42
43
44
45
46
47
48
def delete_item(self, keys: object) -> None:
    """Delete an item from the table by its primary key.

    Args:
        keys: A mapping of key attribute names to their values.
    """
    self._table.delete_item(Key=keys)

get_item(item_id)

Retrieve an item by its ID.

Parameters:

Name Type Description Default
item_id str

The unique identifier of the item to retrieve.

required

Returns:

Type Description
dict | None

The item as a dictionary, or None if not found.

Source code in templates/repository.py
15
16
17
18
19
20
21
22
23
24
def get_item(self, item_id: str) -> dict | None:
    """Retrieve an item by its ID.

    Args:
        item_id: The unique identifier of the item to retrieve.

    Returns:
        The item as a dictionary, or `None` if not found.
    """
    return self._table.get_item(Key={"id": item_id}).get("Item")

list_items()

Retrieve all items from the table.

Returns:

Type Description
list[dict]

A list of all items in the table.

Source code in templates/repository.py
26
27
28
29
30
31
32
def list_items(self) -> list[dict]:
    """Retrieve all items from the table.

    Returns:
        A list of all items in the table.
    """
    return self._table.scan().get("Items", [])

put_item(item)

Write an item to the table, replacing any existing item with the same key.

Parameters:

Name Type Description Default
item dict

A dictionary representing the item to store.

required
Source code in templates/repository.py
34
35
36
37
38
39
40
def put_item(self, item: dict) -> None:
    """Write an item to the table, replacing any existing item with the same key.

    Args:
        item: A dictionary representing the item to store.
    """
    self._table.put_item(Item=item)