API Reference

Auto-generated from source docstrings. Use the source links to view implementation details.

`access_py_telemetry.api`

class access_py_telemetry.api.ApiHandler(*args: Any, **kwargs: Any)[source]

Bases: object

Singleton class to handle API requests. I’m only using a class here so we can save the extra_fields attribute.

Singleton so that we can add extra fields elsewhere in the code and have them persist across all telemetry calls.

To configure request timeouts and the multiprocessing context manually, configure the _request_timeout and _mproc_override class attributes as desired.

add_extra_fields(service_name: str, fields: dict[str, Any]) → None[source]: Add an extra field to the telemetry data. Only works for services that already have an endpoint defined.

clear_headers(service_names: str | Iterable[str] | None = None) → None[source]

Clear the headers for a given service or services, if specified.

If service_names is None, the headers will be cleared for all services.

endpoints = {'intake_catalog': 'intake/catalog', 'intake_cli': 'intake/cli', 'intake_datastore': 'intake/datastore'}

property extra_fields: dict[str, Any]

headers: dict[str, dict[str, str]] = {'intake_catalog': {}, 'intake_cli': {}, 'intake_datastore': {}}

property pop_fields: dict[str, list[str]]

remove_fields(service: str, fields: str | Iterable[str]) → None[source]

Set the fields to remove from the telemetry data for a given service. Useful for excluding default fields that are not needed for a particular telemetry call: eg, removing Session tracking if a CLI is being used.

Note: This does not use a set union, so you must specify all fields you want to remove in one call. # TODO: Maybe make this easier to use?

property request_timeout: float | None

send_api_request(service_name: str, function_name: str, args: list[Any] | tuple[Any, ...], kwargs: dict[str, str | Any]) → None[source]

Send an API request with telemetry data.

Parameters:

service_name (str) – The name of the service to send the telemetry data to.
function_name (str) – The name of the function being tracked.
args (list) – The list of positional arguments passed to the function.
kwargs (dict) – The dictionary of keyword arguments passed to the function.

Return type:

None

Warning

RuntimeWarning: If the request fails.

send_failure_api_request(service_name: str, code: str, endpoint: str) → None[source]

Send an API request with telemetry data, for instance where telemetry has failed parsing and we just want to dump the raw code so it can be interrogated later.

Parameters:

service_name (str) – The name of the service to send the telemetry data to.
code (str) – The code that failed to parse, or otherwise caused an error.
endpoint (str) – The endpoint to send the telemetry data to. This is separate from the service name because we don’t parse a registry of functions for this use case, so we have to do things slightly differently. TODO: I think I built in a side mechanism to do this - find it.

Return type:

None

Warning

RuntimeWarning: If the request fails.

property server_url: str

set_headers(service_names: str | Iterable[str] | None, headers: dict[str, str]) → None[source]

Add headers to the telemetry request for a given service or services, if specified.

If service_names is None, the headers will be added to all services.

class access_py_telemetry.api.ProductionToggle[source]

Bases: object

Singleton class to hold info about whether the code is running in production or not.

This class is a singleton so that the production status can be set once and accessed from anywhere in the code.

Exposed functionality:

production (bool): Whether the code is running in production or not. Setting this will also set the server URL to the production or staging URL.
debug (Callable): A decorator that wraps a function in a try/except block. If the code is running in production, the function will be called normally. Exceptions are suppressed in production but raised in staging.

PRODUCTION_URL = 'https://reporting.access-nri-store.cloud.edu.au/api/'

STAGING_URL = 'https://reporting-dev.access-nri-store.cloud.edu.au/api/'

debug() → Callable[[...], Any][source]

Debugging decorator. Applying this to a function will wrap all telemetry calls in try/except blocks so that users never see any exceptions from the telemetry code.

Notes

We have to apply the branching logic within the decorator, because otherwise the logic gets applied at initialization time, and we can’t change the production status after that.

property production: bool

class access_py_telemetry.api.SessionID[source]

Bases: object

Singleton class to store and generate a unique session ID.

This class ensures that only one instance of the session ID exists. The session ID is generated the first time it is accessed and is represented as a string, using the UUID4 algorithm.

static create_session_id() → str[source]: Generate a unique session ID.

access_py_telemetry.api.send_in_loop(endpoint: str, telemetry_data: dict[str, Any], telemetry_headers: dict[str, str] | None = None, timeout: float | None = None, mproc_override: str | None = None) → None[source]

Wraps the send_telemetry function in an event loop.

If an event loop is already running, sends the telemetry data as a background task.
If no event loop is running, creates a new event loop in a separate process and sends the telemetry data in the background using that loop.

Parameters:

endpoint (str) – The URL to send the telemetry data to.
telemetry_data (dict) – The telemetry data to send.
headers (dict, optional) – The headers to send the telemetry data with.
timeout (float, optional) – The maximum time to wait for the coroutine to finish. If the coroutine takes longer than this time, a TimeoutError will be raised. If None, the coroutine will terminate after 60 seconds. Timeout will also revert to 60 seconds if set to 0.
mproc_override (str, optional) – The multiprocessing context to use. If None, the context will be set to “fork” on Linux systems and “spawn” on Windows/ MacOS systems. If a context is specified, it will be used regardless of the system.

Return type:

None

async access_py_telemetry.api.send_telemetry(endpoint: str, data: dict[str, Any], headers: dict[str, str], warn: bool | None = None) → None[source]

Asynchronously send telemetry data to the specified endpoint.

Parameters:

endpoint (str) – The URL to send the telemetry data to.
data (dict) – The telemetry data to send.
headers (dict) – The headers to send the telemetry data with.
warn (bool, optional) – If True, a warning will be raised if the request fails. If False, no warning will be raised. If None, warn will default the value of ` not ProductionToggle().production`. It wil also enable some status info about the request being sent.

Return type:

None

Warning

RuntimeWarning: If the request fails.

`access_py_telemetry.registry`

exception access_py_telemetry.registry.RegisterWarning[source]

Bases: UserWarning

Custom warning class for the TelemetryRegister class.

class access_py_telemetry.registry.TelemetryRegister(service: str)[source]

Bases: object

Singleton class to register functions for telemetry. Like the session handler, this class is going to be a singleton so that we can register functions to it from anywhere and have them persist across all telemetry calls.

deregister(*func_names: str | Callable[[...], Any]) → None[source]

Deregister a function from the telemetry register.

Parameters:: function_name (str) – The name of the function to deregister. Can also be a list of function names. If you pass a function, it will deregister the function by name, using the __name__ attribute.
Return type:: None

register(*func_names: str | Callable[[...], Any]) → None[source]

Register functions to the telemetry register.

Parameters:: func_names (Sequence[str | Callable]) – The name of the function to register. Can also be a list of function names. If you pass a function, it will register the function by name, using the __name__ attribute.
Return type:: None

`access_py_telemetry.decorators`

access_py_telemetry.decorators.ipy_register_func(service: str, extra_fields: dict[str, Any] | None = None, pop_fields: Iterable[str] | None = None) → Callable[[...], Any][source]

Decorator to register a function in the specified service and track usage using IPython events. This hides a lot of complexity which is more visible in the register_func decorator.

Parameters:

func (Callable) – The function to register.
service (str) – The name of the telemetry register to use.
extra_fields (Iterable[dict[str, Any]], optional) – Extra fields to add to the telemetry record. These can also be added after the fact using the add_extra_field method.
pop_fields (Iterable[str], optional) – Fields to remove from the telemetry record. This can be useful for removing default fields that are not needed for a particular function

Returns:

The function with the telemetry decorator.

Return type:

Callable

access_py_telemetry.decorators.register_func(service: str, extra_fields: dict[str, Any] | None = None, pop_fields: Iterable[str] | None = None) → Callable[[...], Any][source]

Decorator to register a function in the specified service and track usage with async requests.

Parameters:

func (Callable) – The function to register.
service (str) – The name of the telemetry register to use.
fields (extra) – Extra fields to add to the telemetry record. These can also be added after the fact using the add_extra_field method.
pop_fields (Iterable[str], optional) – Fields to remove from the telemetry record. This can be useful for removing default fields that are not needed for a particular function.

Returns:

The function with the telemetry decorator.

Return type:

Callable

`access_py_telemetry.ast`

class access_py_telemetry.ast.CallListener(user_namespace: dict[str, Any], registries: dict[str, set[str]], api_handler: ApiHandler)[source]

Bases: CSTVisitor

METADATA_DEPENDENCIES = (<class 'libcst.metadata.parent_node_provider.ParentNodeProvider'>,): The set of metadata dependencies declared by this class.

visit_Attribute(node: Attribute) → None[source]

visit_Call(node: Call) → None[source]: Visit a call node, process it if it’s a registered call

class access_py_telemetry.ast.ChainSimplifier(user_namespace: dict[str, Any], registries: dict[str, set[str]], api_handler: ApiHandler)[source]

Bases: CSTTransformer

Transform chained calls by removing intermediate method calls Example: ds.search(…).search(…).to_dataset_dict() becomes: ds.to_dataset_dict()

leave_Assign(original_node: Assign, updated_node: Assign) → Assign[source]: When we leave an assignment node, if the value is a call to a registered function, we infer the type of the variable being assigned to. We also handle the case of assigning a variable to another variable, so we can track type information through simple variable assignments. This allows us to resolve the type of variables that are assigned from API calls, and use that type information to simplify chained calls.

leave_Attribute(original_node: Attribute, updated_node: Attribute) → Attribute[source]: When we leave an attribute node, if it’s parent is a cst.Name (ie. the root of a chain of attribute accesses), we replace the value of the attribute with the type name of the instance.

leave_Call(original_node: Call, updated_node: Call) → Call | Name[source]

leave_Subscript(original_node: Subscript, updated_node: Subscript) → Call | Name[source]: When we leave a subscript node, replace eg. instance[key] with ClassName.__getitem__(key). This means there is no need for a CallListener.visit_Subscript method.

access_py_telemetry.ast.capture_registered_calls(info: ExecutionInfo) → None[source]

Use the AST module to parse the code that we are executing & send an API call if we detect specific function or method calls.

Fail silently if we can’t parse the code.

Parameters:: info (IPython.core.interactiveshell.ExecutionInfo) – An object containing information about the code being executed.
Return type:: None

access_py_telemetry.ast.extract_call_args_kwargs(node: Call, user_ns: dict[str, Any]) → tuple[list[Any], dict[str, Any]][source]

Take a cst Call Node and extract the args and kwargs, into a tuple of (args, kwargs)

# TODO: This matcher is a mess, and lacks test coverage. - Add support for f-strings

access_py_telemetry.ast.strip_magic(code: str) → str[source]

Parse the provided code into an AST (Abstract Syntax Tree).

Parameters:: code (str) – The code to parse.
Returns:: The code without IPython magic commands.
Return type:: str

`access_py_telemetry.cli`

Console script for access_py_telemetry.

access_py_telemetry.cli.configure_telemetry(argv: Sequence[str] | None = None) → int[source]: Console script for configuring ipython telemetry.

API Reference

access_py_telemetry.api

access_py_telemetry.registry

access_py_telemetry.decorators

access_py_telemetry.ast

access_py_telemetry.cli

`access_py_telemetry.api`

`access_py_telemetry.registry`

`access_py_telemetry.decorators`

`access_py_telemetry.ast`

`access_py_telemetry.cli`