Tracking

Reference on the Tracking/Telemetry API. Rather, you should use this throug/in conjunction with burr.core.application.ApplicationBuilder.with_tracker().

class burr.tracking.LocalTrackingClient(
project: str,
storage_dir: str = '~/.burr',
serde_kwargs: Dict[str, Any] | None = None,
)

Tracker to track locally – goes along with the Burr UI. Writes down the following: #. The whole application + debugging information (e.g. source code) to a file #. A line for the start/end of each step

__init__(
project: str,
storage_dir: str = '~/.burr',
serde_kwargs: Dict[str, Any] | None = None,
)

Instantiates a local tracking client. This will create the following directories, if they don’t exist: #. The base directory (defaults to ~/.burr) #. The project directory (defaults to ~/.burr/<project>) #. The application directory (defaults to ~/.burr/<project>/<app_id>) on each

On application create, it will write the state machine to the application directory. On pre/post run step, it will write the start/end of each step to the application directory.

Parameters:

  • project – Project name – if this already exists it will be used, otherwise it will be created.

  • storage_dir – Storage directory

classmethod app_log_exists(project: str, app_id: str, storage_dir: str = '~/.burr') → bool

Function to check if state exists for a given project and app_id.

Parameters:

  • project – the name of the project

  • app_id – the application instance id

  • storage_dir – the storage directory.

Returns:

True if state exists, False otherwise.

copy() → LocalTrackingClient

Clones the tracking client. This is useful for forking applications. Note we have to copy the tracking client as it is stateful for tracking. We may make this called internally if it has started already, or make it so it carries multiple states at once.

Returns:

a copy of the self.

list_app_ids(partition_key: str, **kwargs) → list[str]

Returns list of app IDs for a given primary key

load(
partition_key: str,
app_id: str | None,
sequence_id: int | None = None,
**kwargs,
) → PersistedStateData | None

Loads the state for a given app_id

Parameters:

  • partition_key – the partition key. Note this could be None, but it’s up to the persistor to whether that is a valid value it can handle.

  • app_id – the identifier for the app instance being recorded.

  • sequence_id – optional, the state corresponding to a specific point in time. If not provided, should return the latest.

Returns:

position, state, sequence_id

classmethod load_state(
project: str,
app_id: str,
sequence_id: int = -1,
storage_dir: str = '~/.burr',
) → tuple[dict, str]

THis is deprecated and will be removed when we migrate over demos. Do not use! Instead use the persistence API initialize_from to load state.

It defaults to loading the last state, but you can supply a sequence number.

This is a temporary solution – not particularly ergonomic, and makes assumptions (particularly that all logging is in order), which is fine for now. We will be improving this and making it a first-class citizen.

Parameters:

  • project – the name of the project

  • app_id – the application instance id

  • sequence_id – the sequence number of the state to load. Defaults to last index (i.e. -1).

  • storage_dir – the storage directory.

Returns:

the state as a dictionary, and the entry point as a string.

post_application_create(
*,
app_id: str,
partition_key: str | None,
state: State,
application_graph: ApplicationGraph,
parent_pointer: ParentPointer | None,
spawning_parent_pointer: ParentPointer | None,
**future_kwargs: Any,
)

Runs after an “application” object is instantiated. This is run by the Application, in its constructor, as the last step.

Parameters:

  • app_id – Application ID

  • partition_key – Partition key of application

  • state – Current state of the application

  • application_graph – Application graph of the application, representing the state machine

  • parent_pointer – Forking parent pointer of the application (application that it copied from)

  • spawning_parent_pointer – Spawning parent pointer of the application (application that it was launched from)

  • future_kwargs – Future keyword arguments for backwards compatibility

post_run_step(
state: State,
action: Action,
result: dict | None,
sequence_id: int,
exception: Exception,
**future_kwargs: Any,
)

Run after a step is executed.

Parameters:

  • state – State after step execution

  • action – Action that was executed

  • result – Result of the action

  • sequence_id – Sequence ID of the action

  • exception – Exception that was raised

  • future_kwargs – Future keyword arguments

pre_run_step(
state: State,
action: Action,
inputs: Dict[str, Any],
sequence_id: int,
**future_kwargs: Any,
)

Run before a step is executed.

Parameters:

  • state – State prior to step execution

  • action – Action to be executed

  • inputs – Inputs to the action

  • sequence_id – Sequence ID of the action

  • future_kwargs – Future keyword arguments