Skip to content

prefect_aws.utilities

Utilities for working with AWS services.

assemble_document_for_patches(patches)

Assembles an initial document that can successfully accept the given JSON Patch operations.

Parameters:

Name Type Description Default
patches

A list of JSON Patch operations.

 required

Returns:

Type Description

An initial document structured to accept the patches.

 
patches = [
    {"op": "replace", "path": "/name", "value": "Jane"},
    {"op": "add", "path": "/contact/address", "value": "123 Main St"},
    {"op": "remove", "path": "/age"}
]

initial_document = assemble_document_for_patches(patches)

#output
{
    "name": {},
    "contact": {},
    "age": {}
}
Source code in prefect_aws/utilities.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def assemble_document_for_patches(patches):
    """
    Assembles an initial document that can successfully accept the given JSON Patch
    operations.

    Args:
        patches: A list of JSON Patch operations.

    Returns:
        An initial document structured to accept the patches.

    Example:

    ```python
    patches = [
        {"op": "replace", "path": "/name", "value": "Jane"},
        {"op": "add", "path": "/contact/address", "value": "123 Main St"},
        {"op": "remove", "path": "/age"}
    ]

    initial_document = assemble_document_for_patches(patches)

    #output
    {
        "name": {},
        "contact": {},
        "age": {}
    }
    ```
    """
    document = {}

    for patch in patches:
        operation = patch["op"]
        path = patch["path"].lstrip("/").split("/")

        if operation == "add":
            # Ensure all but the last element of the path exists
            ensure_path_exists(document, path[:-1])
        elif operation in ["remove", "replace"]:
            # For remove and replace, the entire path should exist
            ensure_path_exists(document, path)

    return document

ensure_path_exists(doc, path)

Ensures the path exists in the document, creating empty dictionaries or lists as needed.

Parameters:

Name Type Description Default
doc Union[Dict, List]

The current level of the document or sub-document.

 required
path List[str]

The remaining path parts to ensure exist.

 required
Source code in prefect_aws/utilities.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def ensure_path_exists(doc: Union[Dict, List], path: List[str]):
    """
    Ensures the path exists in the document, creating empty dictionaries or lists as
    needed.

    Args:
        doc: The current level of the document or sub-document.
        path: The remaining path parts to ensure exist.
    """
    if not path:
        return
    current_path = path.pop(0)
    # Check if the next path part exists and is a digit
    next_path_is_digit = path and path[0].isdigit()

    # Determine if the current path is for an array or an object
    if isinstance(doc, list):  # Path is for an array index
        current_path = int(current_path)
        # Ensure the current level of the document is a list and long enough

        while len(doc) <= current_path:
            doc.append({})
        next_level = doc[current_path]
    else:  # Path is for an object
        if current_path not in doc or (
            next_path_is_digit and not isinstance(doc.get(current_path), list)
        ):
            doc[current_path] = [] if next_path_is_digit else {}
        next_level = doc[current_path]

    ensure_path_exists(next_level, path)

hash_collection(collection)

Use visit_collection to transform and hash a collection.

Parameters:

Name Type Description Default
collection Any

The collection to hash.

 required

Returns:

Name Type Description
int int

The hash of the transformed collection.
Example 
from prefect_aws.utilities import hash_collection

hash_collection({"a": 1, "b": 2})
Source code in prefect_aws/utilities.py 
 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
def hash_collection(collection) -> int:
    """Use visit_collection to transform and hash a collection.

    Args:
        collection (Any): The collection to hash.

    Returns:
        int: The hash of the transformed collection.

    Example:
        ```python
        from prefect_aws.utilities import hash_collection

        hash_collection({"a": 1, "b": 2})
        ```

    """

    def make_hashable(item):
        """Make an item hashable by converting it to a tuple."""
        if isinstance(item, dict):
            return tuple(sorted((k, make_hashable(v)) for k, v in item.items()))
        elif isinstance(item, list):
            return tuple(make_hashable(v) for v in item)
        return item

    hashable_collection = visit_collection(
        collection, visit_fn=make_hashable, return_data=True
    )
    return hash(hashable_collection)