carlogtt_python_library.database.database_dynamo module

This module …

class carlogtt_python_library.database.database_dynamo.DynamoDB(aws_region_name: str, *, aws_profile_name: str | None = None, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, aws_session_token: str | None = None, caching: bool = False, client_parameters: dict[str, Any] | None = None)[source]

Bases: AwsServiceBase[DynamoDBClient]

The DynamoDB class provides a simplified interface for interacting with Amazon DynamoDB services within a Python application.

It includes an option to cache the client session to minimize the number of AWS API call.

Parameters:

  • aws_region_name – The name of the AWS region where the service is to be used. This parameter is required to configure the AWS client.

  • aws_profile_name – The name of the AWS profile to use for credentials. This is useful if you have multiple profiles configured in your AWS credentials file. Default is None, which means the default profile or environment variables will be used if not provided.

  • aws_access_key_id – The AWS access key ID for programmatically accessing AWS services. This parameter is optional and only needed if not using a profile from the AWS credentials file.

  • aws_secret_access_key – The AWS secret access key corresponding to the provided access key ID. Like the access key ID, this parameter is optional and only needed if not using a profile.

  • aws_session_token – The AWS temporary session token corresponding to the provided access key ID. Like the access key ID, this parameter is optional and only needed if not using a profile.

  • caching – Determines whether to enable caching for the client session. If set to True, the client session will be cached to improve performance and reduce the number of API calls. Default is False.

  • client_parameters – A key-value pair object of parameters that will be passed to the low-level service client.

atomic_writes(put: Iterable[dict[str, str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None]] | None = None, update: Iterable[dict[str, str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None]] | None = None, upsert: Iterable[dict[str, str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None]] | None = None, delete: Iterable[dict[str, str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None]] | None = None, condition_check: Iterable[dict[str, str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None]] | None = None, **kwargs) → dict[str, list[dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None]]][source]

A synchronous write operation that groups up to 100 action requests. These actions can target items in different tables. The actions are completed atomically so that either all of them succeed, or all of them fail.

Parameters:

  • put – Initiates a PutItem operation to write a new item. schema = { ‘TableName’: “string DynamoDB Table Name”, ‘PartitionKeyKey’: “string of the PartitionKey key”, ‘PartitionKeyValue’: “OPTIONAL - partition key value”, ‘SortKeyKey’: “OPTIONAL - string of the SortKey key”, ‘SortKeyValue’: “OPTIONAL - sort key value”, ‘AutoGeneratePartitionKeyValue’: “OPTIONAL - bool”, ‘Items’: “dict containing all the items to put in DynamoDB”, }

  • update – Initiates an UpdateItem operation to update an existing item. schema = { ‘TableName’: “string DynamoDB Table Name”, ‘PartitionKey’: “The partition key as dict of partition_key {key: value}”, ‘SortKey’: “OPTIONAL - The sort key as dict of sort_key {key: value}”, ‘Items’: “dict containing all the values for items to be updated”, ‘ConditionAttribute’: “OPTIONAL - attribute to matched as dict of attribute_to_match {key: value}”, }

  • upsert – Initiates an UpdateOrInsertItem operation to update an existing item or insert if not in the table. schema = { ‘TableName’: “string DynamoDB Table Name”, ‘PartitionKey’: “The partition key as dict of partition_key {key: value}”, ‘SortKey’: “OPTIONAL - The sort key as dict of sort_key {key: value}”, ‘Items’: “dict containing all the values for items to be updated”, ‘ConditionAttribute’: “OPTIONAL - attribute to matched as dict of attribute_to_match {key: value}”, }

  • delete – Initiates a DeleteItem operation to delete an existing item. schema = { ‘TableName’: “string DynamoDB Table Name”, ‘PartitionKey’: “The partition key as dict of partition_key {key: value}”, ‘SortKey’: “OPTIONAL - The sort key as dict of sort_key {key: value}”, }

  • condition_check – Applies a condition to an item that is not being modified by the transaction. The condition must be satisfied for the transaction to succeed. schema = { ‘TableName’: “string DynamoDB Table Name”, ‘PartitionKey’: “The partition key as dict of partition_key {key: value}”, ‘SortKey’: “OPTIONAL - The sort key as dict of sort_key {key: value}”, }

Returns:

A dictionary with keys ‘Put’, ‘Update’, ‘Delete’, ‘ConditionCheck’, and a list of items writes to the DynamoDB in the same order as they were passed in. schema = { ‘Put’: “list of items writes to DynamoDB”, ‘Update’: “list of items writes to DynamoDB”, ‘Upsert’: “list of items writes to DynamoDB”, ‘Delete’: “list of items writes to DynamoDB”, ‘ConditionCheck’: “list of items writes to DynamoDB”, }

Raises:
delete_item(table: str, partition_key_key: str, partition_key_value: bytes | str | float | Iterable[bytes | str | float], sort_key_key: str | None = None, sort_key_value: bytes | str | float | Iterable[bytes | str | float] | None = None) → list[dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None]][source]

Deletes item(s) in a table by primary key.

If a single partition key value is provided, it deletes the corresponding item. If multiple partition key values are provided, it deletes all the corresponding items.

Parameters:

  • table – DynamoDB table name.

  • partition_key_key – The key of the partition key.

  • partition_key_value – The value or an iterable of values of the partition key of the item or items to delete from DynamoDB.

  • sort_key_key – The key of the sort key.

  • sort_key_value – The value or an iterable of values of the sort key of the item or items to delete from DynamoDB.

Returns:

A list of deleted DynamoDB Items deserialized.

Raises:

DynamoDBError – If deletion fails.

delete_item_att(table: str, partition_key_key: str, partition_key_value: bytes | str | float, attributes_to_delete: Iterable[str], sort_key_key: str | None = None, sort_key_value: bytes | str | float | None = None) → dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None][source]

Deletes item specific values in a table by primary key.

Parameters:

  • table – DynamoDB table name.

  • partition_key_key – The key of the partition key.

  • partition_key_value – The value of the partition key.

  • sort_key_key – The key of the sort key.

  • sort_key_value – The value of the sort key.

  • attributes_to_delete – An iterable of specific attributes that are to be deleted from DynamoDB.

Returns:

The updated DynamoDB Item deserialized.

Raises:

DynamoDBError – If deletion fails.

get_item(table: str, partition_key_key: str, partition_key_value: bytes | str | float, sort_key_key: str | None = None, sort_key_value: bytes | str | float | None = None) → dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None] | None[source]

The get_item_from_table operation returns a dictionary of deserialized attribute values for the item with the given primary key. If there is no matching item, get_item_from_table returns None.

Parameters:

  • table – DynamoDB table name.

  • partition_key_key – The key of the partition key.

  • partition_key_value – The value of the partition key.:

  • sort_key_key – The key of the sort key.

  • sort_key_value – The value of the sort key.

Returns:

Deserialized item or None.

Raises:

DynamoDBError – If retrieval fails.

get_items(table: str) → Generator[dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None], None, None][source]

Returns an Iterable of deserialized items in the table.

Parameters:

table – DynamoDB table name.

Returns:

Generator of deserialized items. Iterable of dictionaries of all the columns in DynamoDB i.e. {dynamodb_column_name: column_value, …}

Raises:

DynamoDBError – If retrieval fails.

get_items_count(table: str) → int[source]

Returns the number of items in a table.

Parameters:

table – DynamoDB table name.

Returns:

Item count.

Raises:

DynamoDBError – If count fails.

get_tables() → list[str][source]

Returns an array of table names associated with the current account and endpoint.

Returns:

List of table names.

Raises:

DynamoDBError – If listing fails.

put_atomic_counter(table: str)[source]

In Amazon DynamoDB, there isn’t an in-built auto-increment functionality like in SQL databases for generating record IDs (Primary Key values). However, we can achieve a similar outcome by managing an atomic counter. This method put the initial __PK_VALUE_COUNTER__ item in the table and set the value to 0. If the __PK_VALUE_COUNTER__ item already exists in the table then it does nothing.

Parameters:

table – DynamoDB table name.

Returns:

None

Raises:

DynamoDBError – If operation fails.

put_item(table: str, partition_key_key: str, partition_key_value: bytes | str | float | None = None, sort_key_key: str | None = None, sort_key_value: bytes | str | float | None = None, auto_generate_partition_key_value: bool | None = False, **items: str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None) → dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None][source]

Creates a new item. If an item that has the same primary key as the new item already exists in the specified table, the operation will fail.

Parameters:

  • table – DynamoDB table name.

  • partition_key_key – The key of the partition key.

  • partition_key_value – The value of the partition key.

  • sort_key_key – The key of the sort key.

  • sort_key_value – The value of the sort key.

  • auto_generate_partition_key_value – If set to True, this option instructs DynamoDB to automatically generate a partition key value based on a counter mechanism. For this to work, your table must contain a special item with its partition key value set to ‘__PK_VALUE_COUNTER__’. This item should have a numerical attribute named ‘current_counter_value’, which will be used and incremented as the basis for generating new partition key values.

  • items – Additional items to add.

Returns:

The stored DynamoDB Item deserialized.

Raises:
update_item(table: str, partition_key: dict[str, bytes | str | float], sort_key: dict[str, bytes | str | float] | None = None, condition_attribute: dict[str, Any] | None = None, **items: str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None) → dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None][source]

Performs a strict update on an existing item in DynamoDB.

This method enforces that the item with the specified partition key must already exist before updating. If the item does not exist, or if any specified condition does not match, the update fails with a DynamoDBConflictError. No new item is created in this scenario.

Edits an existing item’s attributes. If condition_attribute_value is passed, the item will be updated only if condition_attribute_value is a match with the value stored in DynamoDB.

Parameters:

  • table – DynamoDB table name.

  • partition_key – DynamoDB partition key as dict of partition_key {key: value}.

  • sort_key – DynamoDB sort key as dict of sort_key {key: value}.

  • condition_attribute – DynamoDB attribute to matched as dict of attribute_to_match {key: value}. When sent to DynamoDB, the attribute will be as a condition to match.

  • items – Values for items to be updated.

Returns:

The updated DynamoDB Item deserialized.

Raises:
upsert_item(table: str, partition_key: dict[str, bytes | str | float], sort_key: dict[str, bytes | str | float] | None = None, condition_attribute: dict[str, Any] | None = None, **items: str | bytes | bytearray | int | float | Decimal | set[str] | set[bytes] | set[int] | set[float] | set[Decimal] | Sequence[Any] | Mapping[str, Any] | bool | None) → dict[str, str | bytes | bytearray | int | float | set[str] | set[bytes] | set[int] | set[float] | Sequence[Any] | Mapping[str, Any] | bool | None][source]

Upsert-like operation.

  1. Tries to update an existing item (strict update).

  2. If that update fails because the item does not exist, it

    puts a brand-new item.

Parameters:

  • table – DynamoDB table name.

  • partition_key – DynamoDB partition key as dict of partition_key {key: value}.

  • sort_key – DynamoDB sort key as dict of sort_key {key: value}.

  • condition_attribute – DynamoDB attribute to matched as dict of attribute_to_match {key: value}. When sent to DynamoDB, the attribute will be as a condition to match.

  • items – Values for items to be updated.

Returns:

The updated DynamoDB Item deserialized.

Raises: