quri_parts.core.utils.recording module#

class RecordLevel(value)#

Bases: IntEnum

Level of recording, which specifies importance of a recording event.

A larger value means higher importance. Record level is a concept similar to logging level. Currently each record level has its counterpart logging level with the same integer value.

INFO = 20#
DEBUG = 10#
INFO = RecordLevel.INFO#

INFO level

DEBUG = RecordLevel.DEBUG#

DEBUG level

class RecordableFunctionId(module, qualname, param)#

Bases: NamedTuple

Represents an identifier for a recordable function.

Parameters:

  • module (str) –

  • qualname (str) –

  • param (Hashable) –

module: str#

Name of the module which the function belongs to.

qualname: str#

Qualified name of the function.

param: Hashable#

Other parameters necessary for identifying a function. It is currently unused.

to_str(full=True)#

Returns a string representation of itself.

If full is True, the returned string contains the module name.

Parameters:

full (bool) –

Return type:

str

class RecordableFunction(*args, **kwargs)#

Bases: Protocol[P, R]

Represents an instance of a recordable function with its identifier, which can be accessed via id attribute.

id: RecordableFunctionId#
recordable(f)#

A decorator for creating a recordable function.

A function to which this decorator is applied must receive a Recorder as its first positional argument, which can be used for recording in the function body. This decorator removes the Recorder argument, so a user of the recordable function does not need to pass a Recorder instance. This decorator also adds a RecordableFunctionId, which can be accesed via id attribute.

Note that when you store mutable data such as list, RecordEntry does not store the snapshot of the data. This means that the data you get is the latest one when you access it. If you want to get the snapshot of the data, you need to copy it by yourself.

Parameters:

f (Callable[[Recorder, P], R]) –

Return type:

RecordableFunction[P, R]

class Recorder(fid)#

Bases: object

Data recorder given to the function which uses data to record.

For the function generated by recordable(), each function call starts with calling start_func(), which creates a new RecordGroup for the function. Note that you should first create RecordSession to record the data.

Parameters:

fid (RecordableFunctionId) –

start_func()#

Context manager to be called for each RecordableFunction call.

Return type:

Iterator[None]

record(level, key, value)#

Records the given data to RecordGroups which belong to active RecordSessions.

Parameters:

  • level (RecordLevel) –

  • key (core.utils.recording._RecKey) –

  • value (core.utils.recording._RecValue) –

Return type:

None

debug(key, value)#

Records the given data with DEBUG level.

Parameters:

  • key (core.utils.recording._RecKey) –

  • value (core.utils.recording._RecValue) –

Return type:

None

info(key, value)#

Records the given data with INFO level.

Parameters:

  • key (core.utils.recording._RecKey) –

  • value (core.utils.recording._RecValue) –

Return type:

None

is_enabled_for(level)#

Checks if there is any active RecordSession which records the data with level or lower.

Parameters:

level (RecordLevel) –

Return type:

bool

class RecordEntry(level, func_id, data)#

Bases: object

Represents a record data entry with its RecordLevel and RecordableFunctionId.

Parameters:
level: RecordLevel#
func_id: RecordableFunctionId#
data: _RecData#
class RecordGroup(func_id, entries, id=<factory>)#

Bases: object

Represents a group of data, which contains the list of RecordEntrys with RecordableFunctionId. This group is created for every RecordableFunction calls.

Parameters:
func_id: RecordableFunctionId#
entries: list[RecordEntry]#
id: int#
add_entry(entry)#

Adds entry to the group.

Parameters:

entry (RecordEntry) –

Return type:

None

class RecordSet#

Bases: object

Set of RecordGroup stored in sequence.

add_group(fid)#

Creates and adds a RecordGroup for given RecordableFunctionId.

Parameters:

fid (RecordableFunctionId) –

Return type:

RecordGroup

remove_last_group()#

Remove the last group from the sequence of RecordGroups.

Return type:

None

get_history(func)#

Returns the RecordGroups corresponding to the RecordableFunction as an Iterable.

Parameters:

func (RecordableFunction[P, R]) –

Return type:

Iterable[RecordGroup]

class RecordSession#

Bases: object

A session manages data recording from recordable functions.

It internally stores recording data received from recordable functions. It also calls associated loggers when receiving data recording events.

set_level(level, func)#

Set a record level for the specified recordable function in this session.

Parameters:
Return type:

None

is_enabled_for(level, fid)#

Checks if recording of the given level is enabled for the specified recordable function in this session.

Returns true if the record level set for the function is equal to or smaller than given level.

Parameters:
Return type:

bool

handler(level, fid, key, value)#

Handles a data recording event.

Internally, a RecordEntry for the event is created and loggers associated with the session are invoked.

Parameters:
Return type:

None

start()#

Starts the data recording session.

Return type:

Iterator[None]

get_records()#

Returns all the records saved in the session.

Return type:

RecordSet

add_logger(logger=None)#

Connect a logger which logs data recording events received by the session.

Parameters:

logger (Logger | None) –

Return type:

None