Conditions

DynamoDB query condition builders.

---

ConditionBaseTranslator

deserialize_condition classmethod

deserialize_condition(condition_expression, is_key=True)

Convert ConditionBase Expression into ConditionBase.

Parameters:

Name	Type	Description	Default
condition_expression	Union[str, ConditionBaseExpression]	The expression to convert.	required
is_key	bool	Used to infer attribute name if expression is string. Defaults to True.	True

Returns:

Type	Description
ConditionBase	The deserialized ConditionBase.

Source code in src/aibs_informatics_aws_utils/dynamodb/conditions.py

@classmethod
def deserialize_condition(
    cls, condition_expression: Union[str, ConditionBaseExpression], is_key: bool = True
) -> ConditionBase:
    """Convert ConditionBase Expression into ConditionBase.

    Args:
        condition_expression: The expression to convert.
        is_key: Used to infer attribute name if expression is string.
            Defaults to True.

    Returns:
        The deserialized ConditionBase.
    """

    def _deserialize_condition(ce: ConditionBaseExpression) -> ConditionBase:
        ce_key = (ce.format, ce.operator)
        condition_base_cls = cls._CONDITION_BASE_CLASS_LOOKUP[ce_key]
        ce_values: list[Union[AttributeBase, ConditionBase]] = []
        for ce_value in ce.values:
            if isinstance(ce_value, ConditionBaseExpression):
                ce_values.append(_deserialize_condition(ce_value))
            elif isinstance(ce_value, AttributeBaseExpression):
                ce_values.append(cls.deserialize_attribute(ce_value))
            else:
                ce_values.append(ce_value)
        return condition_base_cls(*ce_values)

    if isinstance(condition_expression, str):
        condition_string = ConditionBaseExpressionString(condition_expression)
        condition_expression = condition_string.get_condition_expression(is_key=is_key)
    return _deserialize_condition(condition_expression)

ConditionExpressionComponents dataclass

ConditionExpressionComponents(
    expression,
    expression_attribute_names,
    expression_attribute_values,
)

Bases: ExpressionComponentsBase

fix_collisions

fix_collisions(other)

Modify other expression components such that no attribute name/values collide.

Example

Given the following Expression components:

this = ConditionExpressionComponents(
    "#n0 = :v0", {"#n0": "key1"}, {":v0": {"S": "str_value"}}
)
other = ConditionExpressionComponents(
    "#n0 = :v1", {"#n0": "key2"}, {":v1": {"S": "str_value"}}
)

There is a collision of the attribute name #n0 which would get converted to #n1:

output = ConditionExpressionComponents(
    "#n1 = :v1", {"#n1": "key2"}, {":v1": {"S": "str_value"}}
)

Parameters:

Name	Type	Description	Default
other	ConditionExpressionComponents	The other Expression Condition to modify.	required

Returns:

Type	Description
ConditionExpressionComponents	The other, as a modified non-overlapping Expression Condition.

Source code in src/aibs_informatics_aws_utils/dynamodb/conditions.py

def fix_collisions(
    self, other: "ConditionExpressionComponents"
) -> "ConditionExpressionComponents":
    """Modify other expression components such that no attribute name/values collide.

    Example:
        Given the following Expression components:

        ```python
        this = ConditionExpressionComponents(
            "#n0 = :v0", {"#n0": "key1"}, {":v0": {"S": "str_value"}}
        )
        other = ConditionExpressionComponents(
            "#n0 = :v1", {"#n0": "key2"}, {":v1": {"S": "str_value"}}
        )
        ```

        There is a collision of the attribute name `#n0` which would get converted to `#n1`:

        ```python
        output = ConditionExpressionComponents(
            "#n1 = :v1", {"#n1": "key2"}, {":v1": {"S": "str_value"}}
        )
        ```

    Args:
        other: The other Expression Condition to modify.

    Returns:
        The other, as a modified non-overlapping Expression Condition.
    """

    this_placeholder_names = AttrPlaceholder.sorted(self.expression_attribute_names.keys())
    this_placeholder_values = AttrPlaceholder.sorted(self.expression_attribute_values.keys())

    other_placeholder_names = AttrPlaceholder.sorted(other.expression_attribute_names.keys())
    other_placeholder_values = AttrPlaceholder.sorted(other.expression_attribute_values.keys())

    prefix_max_num_map = AttrPlaceholder.build_prefix_max_number_map(
        *this_placeholder_names,
        *this_placeholder_values,
        *other_placeholder_names,
        *other_placeholder_values,
    )

    # Colliding placeholders
    intersect_names = set(this_placeholder_names).intersection(other_placeholder_names)
    intersect_values = set(this_placeholder_values).intersection(other_placeholder_values)

    # create mapping for placeholder updates
    update_map: Dict[AttrPlaceholder, AttrPlaceholder] = {}

    # Now create new placeholders for colliding placeholders,
    # and add to the placeholder update mapping.
    for placeholder in intersect_names.union(intersect_values):
        prefix_max_num_map[placeholder.prefix] += 1
        update_map[placeholder] = AttrPlaceholder.from_components(
            placeholder.prefix, prefix_max_num_map[placeholder.prefix]
        )

    def _fetch(m: Match) -> AttrPlaceholder:
        ap = AttrPlaceholder(m.group(0))
        return update_map.get(ap, ap)

    # Now build the new expression components object
    return ConditionExpressionComponents(
        expression=re.sub(AttrPlaceholder.regex_pattern, _fetch, other.condition_expression),
        expression_attribute_names={
            update_map.get(AttrPlaceholder(p), p): other.expression_attribute_names[p]
            for p in other.expression_attribute_names
        },
        expression_attribute_values={
            update_map.get(AttrPlaceholder(p), p): other.expression_attribute_values[p]
            for p in other.expression_attribute_values
        },
    )

from_condition classmethod

from_condition(condition, is_key_condition)

Create ConditionExpressionComponents from a ConditionBase.

Parameters:

Name	Type	Description	Default
condition	ConditionBase	The ConditionBase to create expression components from.	required
is_key_condition	bool	If the provided condition is for a query() KeyConditionExpression then this should be True. Otherwise if the condition is for a query() FilterExpression or for a put_item() ConditionExpression then this should be set to False.	required

Returns:

Type	Description
ConditionExpressionComponents	The created ConditionExpressionComponents.

Source code in src/aibs_informatics_aws_utils/dynamodb/conditions.py

@classmethod
def from_condition(
    cls, condition: ConditionBase, is_key_condition: bool
) -> "ConditionExpressionComponents":
    """Create ConditionExpressionComponents from a ConditionBase.

    Args:
        condition: The ConditionBase to create expression components from.
        is_key_condition: If the provided condition is for a `query()`
            KeyConditionExpression then this should be True. Otherwise if the condition is for
            a `query()` FilterExpression or for a `put_item()` ConditionExpression then this
            should be set to False.

    Returns:
        The created ConditionExpressionComponents.
    """
    builder = ConditionExpressionBuilder()
    bce = builder.build_expression(condition, is_key_condition=is_key_condition)

    return ConditionExpressionComponents(
        expression=bce.condition_expression,
        expression_attribute_names=cast(Dict[str, str], bce.attribute_name_placeholders),
        expression_attribute_values=cast(Dict[str, Any], bce.attribute_value_placeholders),
    )

condition_to_str

condition_to_str(condition)

Converts a ConditionBase Boto3 object to its str representation

NOTE: Function should be removed if this PR is merged: https://github.com/boto/boto3/pull/3254

Examples:

condition_to_str(Key("name").eq("new_name") & Attr("description").begins_with("new")) (name = new_name AND begins_with(description, new))

condition_to_str(Attr("description").contains("cool")) contains(description, cool)

condition_to_str(None) None

Source code in src/aibs_informatics_aws_utils/dynamodb/conditions.py

def condition_to_str(condition: Optional[ConditionBase]) -> Optional[str]:
    """Converts a ConditionBase Boto3 object to its str representation

    NOTE: Function should be removed if this PR is merged: https://github.com/boto/boto3/pull/3254

    Examples:

    >>> condition_to_str(Key("name").eq("new_name") & Attr("description").begins_with("new"))
    (name = new_name AND begins_with(description, new))

    >>> condition_to_str(Attr("description").contains("cool"))
    contains(description, cool)

    >>> condition_to_str(None)
    None
    """
    if condition is None:
        return None

    builder = ConditionExpressionBuilder()
    expression = builder.build_expression(condition)

    condition_expression = expression.condition_expression

    for name_placeholder, actual_name in expression.attribute_name_placeholders.items():
        condition_expression = condition_expression.replace(name_placeholder, str(actual_name))

    for value_placeholder, actual_value in expression.attribute_value_placeholders.items():
        condition_expression = condition_expression.replace(value_placeholder, str(actual_value))

    return condition_expression