magic_utils package

Subpackages

• magic_utils.exceptions package
  • Submodules
  • magic_utils.exceptions.registryExceptions module
    • DuplicateKeyError
    • MissingKeyError
    • RegistryError
  • Module contents
• magic_utils.magicLogger package
  • Submodules
  • magic_utils.magicLogger.loggingFormatter module
    • ColoredFormatter
      • ColoredFormatter.format()
    • JsonFormatter
      • JsonFormatter.format()
  • magic_utils.magicLogger.loggingManager module
    • setup_logger()
  • Module contents
• magic_utils.registries package
  • Submodules
  • magic_utils.registries.aliasRegistry module
    • AliasRegistry
      • AliasRegistry.add_alias()
      • AliasRegistry.aliases
      • AliasRegistry.canonicals
      • AliasRegistry.get()
      • AliasRegistry.register()
      • AliasRegistry.register_class()
      • AliasRegistry.register_function()
      • AliasRegistry.remove()
      • AliasRegistry.remove_alias()
      • AliasRegistry.reset()
      • AliasRegistry.resolve_key_to_canonical()
      • AliasRegistry.resolve_value_to_canonicals()
      • AliasRegistry.update()
  • magic_utils.registries.registry module
    • Registry
      • Registry.get()
      • Registry.register()
      • Registry.register_class()
      • Registry.register_function()
      • Registry.registry
      • Registry.remove()
      • Registry.reset()
      • Registry.update()
  • Module contents
• magic_utils.tunnel package
  • Submodules
  • magic_utils.tunnel.tunnelmole module
    • TunnelMole
      • TunnelMole.is_running()
      • TunnelMole.start()
      • TunnelMole.stop()
  • Module contents

Submodules

magic_utils.eventManager module

class magic_utils.eventManager.EventManager

Bases: object

A class to manage subscribers and their notifications.

The EventManager class manages a list of subscribers and supports adding, removing, and notifying these subscribers. Subscribers can be notified sequentially with notify_all or in parallel with notify_parallel.

Variables:

subscribers – A list where each element is a tuple containing a callable subscriber and its associated arguments and keyword arguments.

clear()

Remove all subscribers from the EventManager.

notify_all() → dict[tuple[str, tuple[any], tuple[tuple[str, any]] | tuple], any]

Notify all subscribers sequentially.

Returns:

Dict of {tuple(sub, args, tuple(kwargs.items())) : returned_value}. The order of the items represents the order in which they were called

notify_parallel(num_threads: int = 2) → dict[tuple[str, tuple[any], tuple[tuple[str, any]] | tuple], any]

Notify all subscribers in parallel using the specified number of threads.

If the specified number of threads exceeds the number of subscribers, a warning will be issued and the number of threads will be automatically adjusted to match the number of available subscribers.

Parameters:

num_threads – The number of threads to use for parallel notification. Must be 2 or higher.

Returns:

Dict of {tuple(sub, args, tuple(kwargs.items())) : returned_value}. The order of the items represents the order in which they were called.

Raises:

ValueError – If num_threads is less than 2.

subscribe(subscriber: callable, *subscriber_args, **subscriber_kwargs)

Add a new subscriber to the EventManager.

Parameters:

• subscriber – A callable object that will be notified.

• subscriber_args – Arguments to be passed to the subscriber callable.

• subscriber_kwargs – Keyword arguments to be passed to the subscriber callable.

unsubscribe(subscriber: callable)

Remove a subscriber from the EventManager.

Parameters:

subscriber – The callable object to be removed from the list of subscribers.

Module contents

Magic Utils - A collection of useful utility classes and functions.

class magic_utils.AliasRegistry(registry_name: str = 'BaseRegistry', *args, **kwargs)

Bases: Registry

add_alias(key: Hashable, alias: Hashable) → None

Add an alias to an existing canonical key.

Parameters:

• key – A canonical key or alias.

• alias – The new alias.

Raises:

DuplicateKeyError – If alias already exists.

property aliases: dict[Hashable, Hashable]

Return mapping of alias -> canonical.

property canonicals: dict[Hashable, set[Hashable]]

Return mapping of canonical -> aliases.

get(key: Hashable) → Any

Retrieve a value by key or alias.

Parameters:

key – Canonical key or alias.

register(canonical: Hashable, value: Any, aliases: list[Hashable] | None = None) → None

Register a canonical key with optional aliases.

Ensures that the canonical and all aliases are globally unique across both canonical keys and aliases.

Parameters:

• canonical – The primary key.

• value – Value to store.

• aliases – Optional list of aliases.

Raises:

DuplicateKeyError – On duplicate canonical or alias.

register_class(key: Hashable | None = None, aliases: list[Hashable] | None = None) → object

Decorator to register a class with optional aliases.

Parameters:

• key – Canonical key.

• aliases – Optional aliases.

register_function(key: Hashable | None = None, aliases: list[Hashable] | None = None) → Any

Decorator to register a function with optional aliases.

Parameters:

• key – Canonical key.

• aliases – Optional aliases.

remove(key: Hashable | None = None, value: Any = None, rmv_all: bool = False) → None

Remove an entry and all its aliases.

Parameters:

• key – Key or alias.

• value – Value.

• rmv_all – Passed to base class.

remove_alias(key: Hashable, alias: Hashable) → None

Remove an alias from a canonical key.

Parameters:

• key – A canonical key or alias.

• alias – The alias to remove.

Raises:

MissingKeyError – If key or alias not found.

reset() → None

Clear registry and all aliases.

resolve_key_to_canonical(key: Hashable) → Hashable

Resolve a key or alias to its canonical key.

resolve_value_to_canonicals(value: Any) → set[Hashable]

Resolve a value to its canonical keys.

update(key: Hashable, value: Any) → None

Update a value by key or alias.

Parameters:

• key – Canonical key or alias.

• value – New value.

class magic_utils.EventManager

Bases: object

A class to manage subscribers and their notifications.

The EventManager class manages a list of subscribers and supports adding, removing, and notifying these subscribers. Subscribers can be notified sequentially with notify_all or in parallel with notify_parallel.

Variables:

subscribers – A list where each element is a tuple containing a callable subscriber and its associated arguments and keyword arguments.

clear()

Remove all subscribers from the EventManager.

notify_all() → dict[tuple[str, tuple[any], tuple[tuple[str, any]] | tuple], any]

Notify all subscribers sequentially.

Returns:

Dict of {tuple(sub, args, tuple(kwargs.items())) : returned_value}. The order of the items represents the order in which they were called

notify_parallel(num_threads: int = 2) → dict[tuple[str, tuple[any], tuple[tuple[str, any]] | tuple], any]

Notify all subscribers in parallel using the specified number of threads.

If the specified number of threads exceeds the number of subscribers, a warning will be issued and the number of threads will be automatically adjusted to match the number of available subscribers.

Parameters:

num_threads – The number of threads to use for parallel notification. Must be 2 or higher.

Returns:

Dict of {tuple(sub, args, tuple(kwargs.items())) : returned_value}. The order of the items represents the order in which they were called.

Raises:

ValueError – If num_threads is less than 2.

subscribe(subscriber: callable, *subscriber_args, **subscriber_kwargs)

Add a new subscriber to the EventManager.

Parameters:

• subscriber – A callable object that will be notified.

• subscriber_args – Arguments to be passed to the subscriber callable.

• subscriber_kwargs – Keyword arguments to be passed to the subscriber callable.

unsubscribe(subscriber: callable)

Remove a subscriber from the EventManager.

Parameters:

subscriber – The callable object to be removed from the list of subscribers.

class magic_utils.Registry(registry_name: str = 'BaseRegistry', *args, **kwargs)

Bases: object

get(key: Hashable) → Any | None

Retrieve an item from the registry.

Parameters:

key – The key to retrieve.

Returns:

The value associated with the key.

Raises:

NotRegisteredError – If the key is not registered and exceptions are enabled.

register(key: Hashable, value: Any) → None

Register one key with a specified value. Skips registration if a key is already registered completely.

Parameters:

• key – A hashable key to register.

• value – The value to associate with the given key.

Raises:

DuplicateError – If a key is already registered and exceptions are enabled.

register_class(key: Hashable | None = None)

A decorator to register a class to the registry.

Parameters:

key – The key to register the class with.

Returns:

The decorated class.

register_function(key: Hashable | None = None) → Any

A decorator to register a function itself to the registry. This registration happens only once when the function is first defined.

Parameters:

key – A single key to register the function with.

Returns:

The decorated function.

property registry: Dict[Hashable, Any]

Access the underlying registry.

Returns:

The dictionary containing all registered items.

Return type:

Dict[Hashable, Any]

remove(key: Hashable | None = None, value: Any = None, rmv_all: bool = False) → None

Remove an item from the registry by key or value.

Parameters:

• key – The key to remove. If None, removal is based on the value.

• value – The value to remove. If None, removal is based on the key.

• rmv_all – Whether to remove all matches. If True, removes all matches. Only needed if removed by value.

Else raises error if multiple matches are found. :raises ValueError: If both key and value are passed or None. :raises NotRegisteredError: If the key or value is not found and exceptions are enabled.

reset() → None

Clear all items from the registry.

Returns:

None

update(key: Hashable, value: Any) → None

Update an item in the registry.

Parameters:

• key – The key to update. If None, removal is based on the value.

• value – The value to update to.

class magic_utils.TunnelMole(port: int = 8000)

Bases: object

Manage a tunnelmole process that exposes a local port via a public URL.

is_running() → bool

Check if tunnelmole is still running.

start() → str | None

Start the tunnelmole process.

Returns:

If successful returns None, if no url was found the shell response of the command is returned.

stop()

Stop the tunnelmole process if it’s running.

magic_utils.setup_logger(logger_name: str, log_file_path: str, stream_level: int = 10, log_level: int = 10, stream_in_color: bool = True, stream_formatter: Formatter | Literal['default'] | None = 'default', file_formatter: Formatter | Literal['default'] | None = 'default', log_in_json: bool = True, extra_log_args: list[str] = None, remove_previous_handlers: bool = True, timed_rotating_file_handler_kwargs: dict | None = None) → Logger

Configure and initialize a logger with stream and timed rotating file handlers.

Note:

• logger.propagate = False

• removes previous handlers from logger

• Default format (this isn’t the colored format): [%(asctime)s | %(levelname)s] [%(filename)s | lineno%(lineno)d | %(funcName)s] => %(message)s

Parameters:

• logger_name – The name of the logger to configure.

• log_file_path – The filename where logs will be written, with daily rotation. (e.g., ‘/logs/app.jsonl’)

• stream_level – The log level for the stream handler (e.g., logging.DEBUG, logging.INFO) Defaults to DEBUG.

• log_level – The log level for the file handler (e.g., logging.DEBUG, logging.INFO) Defaults to DEBUG.

• stream_in_color – If True, logs to stdout will use colored formatting. Defaults to True.

• stream_formatter – The formatter to use for the stream handler. Defaults to ColoredFormatter if stream_in_color is True,

otherwise uses the default formatter when ‘default’ is used. To disable the stream Handler set it to None. :param file_formatter: The formatter to use for the file handler. Defaults to JsonFormatter if log_in_json is True, otherwise uses the default formatter when ‘default’ is used. To disable the stream Handler set it to None. :param log_in_json: If True, logs to file will be written in JSON format. Defaults to True. :param remove_previous_handlers: If True, removes previous handlers from the logger. Defaults to True. :param extra_log_args: List of extra attribute keys to include in the logs (e.g., [‘arg1’, ‘arg2’]).

These keys will be added to the colored formatter output if present.

Parameters:

timed_rotating_file_handler_kwargs – Kwargs to pass to the TimedRotatingFileHandler constructor. Overwrites all the default kwargs {‘filename’: log_file_name, ‘when’: ‘midnight’, ‘interval’: 1, ‘backupCount’: 3}

Returns:

None