Skip to main content

Using the client tracking mode

The Python SDK provides a feature known as Tracking Mode. This mode allows for the aggregation and tracking of operations performed during a session, enhancing efficiency and data management.

Overview​

Tracking mode is designed for scenarios where multiple operations or mutations are performed in sequence, and you want to consolidate or track these operations for efficiency or auditing purposes. It is ideal for scenarios requiring precise control and repeatability, such as idempotent scripting. Under the hood, tracking mode leverages a CoreStandardGroup object to aggregate and track these operations.

For a detailed exploration of tracking mode applications and how to use it for idempotent scripting, visit the tracking topic.

Automatic tracking​

The Infrahub Python SDK offers a streamlined process for managing the tracking session lifecycle using the context manager. This method is available for both asynchronous and synchronous clients, ensuring all operations within the context are efficiently tracked without manual intervention.

Utilizing the context manager​

With the start_tracking method, you initiate tracking mode as you enter the context and automatically conclude the tracking session as you exit. This guarantees that all operations performed within the context are tracked under the specified session, simplifying tracking management.

# Auto-manage tracking session with async context manager
async with client.start_tracking(identifier="my_tracking_session", params=params, delete_unused_nodes=True) as session:
    # Tracked operations
    node = await session.create(kind="MyNodeKind", data={"name": "Example"})
    await node.save()

    # Optionally, add related Nodes and Groups to the context
    await session.group_context.add_related_nodes([another_node.id])
    await session.group_context.add_related_groups([group.id])
info

The context manager feature elegantly handles the start and conclusion of tracking sessions, making your code cleaner and less prone to errors related to manual tracking management.

Manual tracking​

Manual tracking involves explicitly starting, managing, and concluding sessions, offering fine-grained control over the tracking process.

Setting up​

First, ensure the Python SDK is installed and configured to communicate with your Infrahub instance. Then, let's start by setting up our client and enabling tracking mode.

Once tracking mode is enabled, it's important to note that only the objects that are being saved (created or updated) will be automatically tracked under the specified session identifier. Objects that are merely queried and not modified or saved will not be added to the tracking group. This behavior ensures that the tracking group specifically reflects changes made during the session, providing a clear audit trail of modifications.

from infrahub_sdk.client import InfrahubClient

client = InfrahubClient(address="http://localhost:8000")

await client.start_tracking(identifier="my_tracking_session")
node = await client.create(kind="MyNodeKind", data={"name": "Example"})
await node.save()

Tracking parameters​

When enabling tracking mode with the start_tracking method, you can customize the tracking session with the following parameters:

ParameterDescription
identifierUnique string to identify the session, used for correlating operations and logs. Defaults to "python-sdk" if not specified.
paramsOptional dictionary for extra context, enabling fine-grained control over tracking.
delete_unused_nodesBoolean indicating if nodes not referenced should be automatically deleted, helping maintain a clean state.
group_typeType of group object for tracking, default is CoreStandardGroup, customizable for specific grouping logic.

These parameters provide flexibility, enabling detailed auditing, efficient data management, and support for idempotent operations.

Advanced tracking with parameters​

In addition to the basic tracking mode functionalities, the Infrahub Python SDK allows for more granular control over the tracking sessions through parameters. These parameters can be beneficial for categorizing, filtering, or adding metadata to the groups created or updated during the tracking session. When starting a tracking session with start_tracking, you can pass a dictionary of parameters to define further the context and characteristics of the group being tracked. This allows for dynamic grouping based on runtime data, user inputs, or other operational metrics.

params = {
    "data_source": "external",
    "analysis_type": "ipam_data"
}

await client.start_tracking(
    identifier="weekly_analysis",
    params=params,
    delete_unused_nodes=True
)

Ending a tracking session and updating tracking information​

This step involves creating or updating (using upsert) the CoreStandardGroup used to store all the Nodes and Groups used during execution.

from infrahub_sdk.client import InfrahubClient

client = InfrahubClient(address="http://localhost:8000")

await client.start_tracking(identifier="my_tracking_session")
node = await client.create(kind="MyNodeKind", data={"name": "Example"})
await node.save()

# Update tracking information for async client
await client.group_context.update_group()

Retrieving and manipulating groups​

After setting up a tracking session with specific parameters, you may need to retrieve the group associated with this session for further manipulation. This involves fetching the group, accessing its members, and executing specific logic based on the member types.

Here's how to retrieve a group based on an identifier and parameters and interact with its members:

# Set the context properties to match the tracking session
await client.set_context_properties(identifier="my_tracking_session", params=params)

# Retrieve the group associated with the specified identifier and parameters
group = await client.group_context.get_group(store_peers=True)

# Check if the group exists
if group:
    # Access previous members of the group, if any
    if client.group_context.previous_members:
        for member in client.group_context.previous_members:
            # Fetch the object from the store based on the member's type and ID
            obj = client.store.get(kind=member._typename, key=member.id)

            # Perform operations based on the member's type
            pass