Transactions

Transaction wrappers for ACID operations across DynamoDB tables.

transactions

Transaction support for Dynantic.

Provides ACID transactions across DynamoDB tables using TransactWriteItems and TransactGetItems.

TransactPut

Wraps a model instance for transactional put.

Parameters:

Name	Type	Description	Default
item	DynamoModel	The DynamoModel instance to save.	required
condition	Condition | None	Optional condition expression that must be satisfied for the put to succeed.	None

Example

TransactPut(user, condition=Attr("user_id").not_exists())

Source code in dynantic/transactions.py

class TransactPut:
    """Wraps a model instance for transactional put.

    Args:
        item: The DynamoModel instance to save.
        condition: Optional condition expression that must be satisfied for the put to succeed.

    Example:
        TransactPut(user, condition=Attr("user_id").not_exists())
    """

    def __init__(self, item: DynamoModel, condition: Condition | None = None) -> None:
        self.item = item
        self.condition = condition

    def _to_transact_item(self) -> dict[str, Any]:
        """Build the DynamoDB TransactWriteItem dict for this put operation."""
        config = self.item._meta
        serializer = self.item._serializer

        data = self.item.model_dump(mode="python", exclude_none=True)

        # Handle TTL conversion
        from .config import convert_ttl_fields

        convert_ttl_fields(data, config)

        dynamo_item = serializer.to_dynamo(data)

        put: dict[str, Any] = {
            "TableName": config.table_name,
            "Item": dynamo_item,
        }

        if self.condition is not None:
            from .conditions import compile_condition

            condition_params = compile_condition(self.condition, serializer)
            put.update(condition_params)

        return {"Put": put}

TransactDelete

Wraps a delete operation for transactional delete.

Parameters:

Name	Type	Description	Default
model_cls	type[DynamoModel]	The DynamoModel class of the item to delete.	required
condition	Condition | None	Optional condition expression that must be satisfied for the delete to succeed.	None
**key_values	Any	Primary key values identifying the item to delete.	{}

Example

TransactDelete(Order, condition=Attr("status") == "cancelled", order_id="o1")

Source code in dynantic/transactions.py

class TransactDelete:
    """Wraps a delete operation for transactional delete.

    Args:
        model_cls: The DynamoModel class of the item to delete.
        condition: Optional condition expression that must be satisfied for the delete to succeed.
        **key_values: Primary key values identifying the item to delete.

    Example:
        TransactDelete(Order, condition=Attr("status") == "cancelled", order_id="o1")
    """

    def __init__(
        self,
        model_cls: type[DynamoModel],
        condition: Condition | None = None,
        **key_values: Any,
    ) -> None:
        self.model_cls = model_cls
        self.key_values = key_values
        self.condition = condition

    def _to_transact_item(self) -> dict[str, Any]:
        """Build the DynamoDB TransactWriteItem dict for this delete operation."""
        config = self.model_cls._meta
        serializer = self.model_cls._serializer

        dynamo_key = serializer.to_dynamo(self.key_values)

        delete: dict[str, Any] = {
            "TableName": config.table_name,
            "Key": dynamo_key,
        }

        if self.condition is not None:
            from .conditions import compile_condition

            condition_params = compile_condition(self.condition, serializer)
            delete.update(condition_params)

        return {"Delete": delete}

TransactConditionCheck

Wraps a condition check for transactional validation without modifying the item.

Parameters:

Name	Type	Description	Default
model_cls	type[DynamoModel]	The DynamoModel class of the item to check.	required
condition	Condition	Condition expression that must be satisfied for the transaction to succeed.	required
**key_values	Any	Primary key values identifying the item to check.	{}

Example

TransactConditionCheck(Account, Attr("balance") >= 100, account_id="acc-1")

Source code in dynantic/transactions.py

class TransactConditionCheck:
    """Wraps a condition check for transactional validation without modifying the item.

    Args:
        model_cls: The DynamoModel class of the item to check.
        condition: Condition expression that must be satisfied for the transaction to succeed.
        **key_values: Primary key values identifying the item to check.

    Example:
        TransactConditionCheck(Account, Attr("balance") >= 100, account_id="acc-1")
    """

    def __init__(
        self,
        model_cls: type[DynamoModel],
        condition: Condition,
        **key_values: Any,
    ) -> None:
        self.model_cls = model_cls
        self.key_values = key_values
        self.condition = condition

    def _to_transact_item(self) -> dict[str, Any]:
        """Build the DynamoDB TransactWriteItem dict for this condition check."""
        from .conditions import compile_condition

        config = self.model_cls._meta
        serializer = self.model_cls._serializer

        dynamo_key = serializer.to_dynamo(self.key_values)

        check: dict[str, Any] = {
            "TableName": config.table_name,
            "Key": dynamo_key,
        }

        condition_params = compile_condition(self.condition, serializer)
        check.update(condition_params)

        return {"ConditionCheck": check}

TransactGet

Wraps a get operation for transactional reads.

Parameters:

Name	Type	Description	Default
model_cls	type[DynamoModel]	The DynamoModel class of the item to read.	required
**key_values	Any	Primary key values identifying the item to read.	{}

Example

TransactGet(User, user_id="u1")

Source code in dynantic/transactions.py

class TransactGet:
    """Wraps a get operation for transactional reads.

    Args:
        model_cls: The DynamoModel class of the item to read.
        **key_values: Primary key values identifying the item to read.

    Example:
        TransactGet(User, user_id="u1")
    """

    def __init__(self, model_cls: type[DynamoModel], **key_values: Any) -> None:
        self.model_cls = model_cls
        self.key_values = key_values

    def _to_transact_item(self) -> dict[str, Any]:
        """Build the DynamoDB TransactGetItem dict for this get operation."""
        config = self.model_cls._meta
        serializer = self.model_cls._serializer

        dynamo_key = serializer.to_dynamo(self.key_values)

        return {
            "Get": {
                "TableName": config.table_name,
                "Key": dynamo_key,
            }
        }