`luminal.managers.loader`

Module Contents

Classes

Loader

Allows management of photons and how they are loaded, unloaded, reloaded, and monitored.

class Loader(logging=False, suppress_errors=False)

Allows management of photons and how they are loaded, unloaded, reloaded, and monitored.

Parameters:

logging (bool) –
suppress_errors (bool) –

property photons: dict[str, luminal.managers.handler.Handler]

Returns a dictionary of photon names and corresponding Handler objects.

Returns:: A dictionary containing photon names as keys and their corresponding Handler objects as values.
Return type:: Dict[str, Handler]

Notes

This is a read-only property, meaning that it can only be accessed and not modified.
The photon name is the name assigned to the photon instance in its definition.
The Handler objects contain information about the photon, including its name, file path, and the IPhoton based class itself.

async _validate_module(imported_module, photon_path, photon_base, other_classes=[])

|coro|

Checks for all photon instances, specified classes and creates handlers objects for them.

Parameters:

imported_module (ModuleType) – The imported module to be validated.
photon_path (str) – The absolute path of the photon module.
photon_base (type) – The base class that all photons of the system will inherit from.
other_classes (Optional[list[str]]) – A list of additional classes to check for in the imported module. Defaults to [].

Returns:

If the module is a valid photon module, a list of Handler objects that correspond to the photon instances and specified classes in the module is returned. If it is not a valid photon module, returns False.

Return type:

Union[List[Handler], bool]

Notes

Even if the module is not a valid photon module and it returns False, the imported_module will still be deleted from the system.

async _import_module(module_path)

|coro|

Resolves, executes, and returns a Python module based on the provided module path.

Parameters:: module_path (str) – The absolute path of the module to be imported.
Returns:: The imported module object.
Return type:: ModuleType
Raises:: ModuleNotFoundError – If the module cannot be found (either due to an incorrect path or it not existing), or if a loader or spec cannot be found for the module.

Notes

The imported module is added to the sys.modules dictionary for future reference, and is also returned.
If an error occurs while importing the module and the logging property is set, the error will be logged.
If suppress_errors is set, the error will be skipped.

async _scan_module(path)

|coro|

Determines if the provided path is a package or module.

Parameters:: path (str) – Absolute path of the file or directory to be checked.
Returns:: A tuple containing the absolute path of the module, a flag indicating whether the path is a package, and a flag indicating if there are only packages in the directory.
Return type:: Tuple[str, bool, bool]
Raises:: ModuleNotFoundError – If the file path provided does not exist, is not accessible or is not a valid photon.

Notes

The first element, path in the tuple returned represents the absolute path of the individual module.
The second element, is_dir in the tuple returned is a boolean value that indicates whether the module is in fact a package.
The third element of the returned tuple, packages_only is a boolean value that indicates whether there are only packages in the directory.

async _find_modules(path)

|coro|

Creates a collection of module paths and determines if they’re packages.

Parameters:: path (str) – The path of the directory containing the modules.
Returns:: A tuple containing a list of tuples representing available modules and if they’re Python packages, and a boolean representing if there are only packages.
Return type:: Tuple[List[Tuple[str, bool]], bool]

Notes

If path is not a valid directory or cannot be accessed, an empty list will be returned.
The first element, module_path in the tuple returned represents the absolute path of the individual module.
The second element, is_dir in the tuple returned is a boolean value that indicates whether the module is in fact a package.
The second element of the returned tuple, packages_only is a boolean value that indicates whether there are only packages in the directory.

async _check_photon(photon)

|coro|

Check if the photon belongs to the available photon registry and return the corresponding handler(s).

Parameters:: photon (str|Handler) – Name of the photon or Handler object.
Returns:: If photon exists in the available photons, return the corresponding handler(s).
Return type:: Union[Handler, List[Handler], None]
Raises:: PhotonNotFoundError: – If photon is not a valid file path or if it does not exist in the available photons.

Notes

If photon is a Handler object, return it as is. If photon does not exist in the available photons, return None.

async _emit_photon(photon_path, photon_base=IPhoton, other_classes=[], recursive=False)

|coro|

Loads and returns a photon from a given path.

Parameters:

photon_path (str | tuple[str, bool, bool]) – A string representing the file path of the photon or a tuple representing the photon path along with boolean values indicating if the photon is a module or a package.
photon_base (Optional[type]) – Base class for photons. Default is IPhoton.
other_classes (Optional[list[str]]) – List of classes that the photon should inherit from. Default is [].
recursive (Optional[bool]) – A flag indicating whether to look for photons recursively in subdirectories under the photons_directory. Default is False.

Returns:

The handler object of the photon that was emitted, a list of handler objects of photons that were emitted, or None.

Return type:

Handler | list[Handler] | None

Raises:

ModuleNotFoundError – When the photon’s module is not found.
Exception – Generally when a photon author has syntax errors or other issues with their photon source.

Notes

The _emit_photon() function first initializes an empty list called photons. It then checks whether the given path is a module file or a package directory by checking if the type of photon_path is a tuple. If so, it retrieves the information about the path’s readiness, the module path, and whether the specified path contains only packages. If the photon is a package and the recursive flag is True or the package_only flag is True, the function reads photons from other levels down in the package.
It then checks if the photon has already been loaded by resolving the module path and accessing the _photons attribute of the instance of Loader. If the photon is already loaded, a warning message is logged, and None is returned.
If the photon is not loaded, then the function imports the photon’s module path using the _import_module() method, and validates the module with the given photon base class and a list of other classes the photon should inherit from using the _validate_module() method.
It returns a list of photons that were validated. If the list of photons is empty, a warning message is logged, and None is returned. Otherwise, a success message is logged for loading the photon, and the photons are added to the list.
Finally, the function returns a list of handler objects for the loaded photons.

Examples

>>> loader = Loader()
>>> photon = await loader._emit_photon('photons/hello_world.py', IPhoton, ["Other_class1", "Other_class2"])

async _emit_photons(photons_directory, photon_base=IPhoton, other_classes=[], recursive=False)

|coro|

Retrieves photon handlers from a specified directory.

Parameters:

photons_directory (str) – A string representing the directory where the photons are located.
photon_base (Optional[type]) – An optional argument representing the base class for the photons. Default is IPhoton.
other_classes (Optional[list[str]]) – List of strings representing other classes that the photons should inherit from. Default is [].
recursive (Optional[bool]) – A boolean indicating whether to look for photons recursively in subdirectories under photons_directory. Default is False.

Returns:

A list of photons representing the handlers stored in the given directory and its subdirectories.

Return type:

list[Handler]

:raises Any exceptions raised by the called _emit_photon() method.:

See also

_emit_photon(): Loads and returns a photon from a given path.

Examples

>>> loader = Loader()
>>> photons = await loader._emit_photons("photons_directory", IPhoton, ["Other_class1", "Other_class2"])

async _absorb_photon(photon, force_stop=False, suppress_finalizer_log=False)

|coro|

Stops the specified photon and removes it from the loader’s photon registry. If the given photon is a string, it is looked up in the registry by filepath. If the photon is not found, this method returns False.

Parameters:

photon (Handler|str) – Either a Handler object that represents a running photon, or a str name of a photon in the photon registry.
force_stop (bool) – A flag that specifies whether to force-stop the photon if it does not stop within a reasonable time or does not implement a finalizer.
suppress_finalizer_log (bool) –

Returns:

Indicating whether the photon was successfully stopped and removed.

Return type:

bool

async _absorb_photons(photons, force_stop=False)

|coro|

Stops and removes a list of photons from the loader’s registry. If a given photon is a string, it is looked up in the registry by filepath. The method returns a list of successful photon filepaths that were stopped and absorbed.

Parameters:

photons (list[Handler|str]) – A list of Handler objects representing photons, or a list of str names of the photons in the photon registry.
force_stop (bool) –

Returns:

A list of successful photon filepaths that were stopped and absorbed.

Return type:

list[str]

async _revert_photon(validated_photon, validated_modules)

|coro|

Rolls back a photon to a working state if an atomic reload fails.

Parameters:

validated_photon (Union[Handler, List[Handler]]) – A single Handler object or a list of Handler objects which are validated successfully during the atomic reload.
validated_modules (Dict[str, ModuleType]) – A dictionary containing the validated modules for the given photon. These modules are updated in the sys.modules after the photon has been rolled back.

Return type:

None

:raises Any exception that may occur during the validated_photon.start() call.:

Notes

This function is responsible for rolling back a photon to a working state if an atomic reload fails. It takes two parameters: a validated photon and validated modules dictionary. If the validation passes successfully, then the _revert_photon() function starts the validated photon by calling the validated_photon.start() function. Afterward, the _photons collection is updated to hold the working photon, and updated modules are passed to the sys.modules by the sys.modules.update() call.

This function expects the provided validated_photon to be a single Handler object or a list of such objects. It is also expected that the provided validated_photon has passed the validation check without any errors.

Please note that any exception that is raised during the await validated_photon.start() function call is propagated back to the caller. The exceptions could arise due to coding bugs, configuration issues, or other environmental reasons.

async _reload_photon(photon)

|coro|

Reloads a photon handler or an entire photon atomically and returns the updated Handler object(s).

Parameters:

photon (Union[Handler, str]) – A single handler object or the string name for the photon which needs to be reloaded.

Returns:

If a single photon is provided, it returns the updated Handler object.
If multiple photons are provided or the photon to be reloaded has multiple handlers,
it returns a list of the updated Handler objects.

Return type:

Union[Handler, List[Handler]]

Raises:

Any exception that may occur during: –

self._check_photon()
_absorb_photon(validated_photon)()
_emit_photon(photon)()
_revert_photon(validated_photon, validated_modules)()

Notes

This function reloads a photon handler or an entire photon atomically and returns the updated Handler object(s). The function starts by validating the provided photon by calling the _check_photon(photon)() function. Once the validation is complete, this function collects all the validated modules for the photon across the system.

Depending on the number of validated photons i.e., whether a single Handler object or a list of them is provided, the function iterates through each of them and collects all validated modules for each. It then calls the _absorb_photon(validated_photon)() function to atomically update the photon(s). After an atomic update of the photon(s), it returns the updated Handler object(s) by calling the _emit_photon(photon)() function.

In case any exception occurs during the atomic update, this function calls the _revert_photon(validated_photon, validated_modules)() function to fall back to the previous working state.

Please note that any exception that is raised during any defined calls in this function is propagated back to the caller. The exceptions may arise due to coding bugs, configuration issues, or other environmental reasons.

async _reload_photons(photons)

|coro|

Reloads a list of photon handlers or entire photons atomically and returns the updated Handler objects.

Parameters:: photons (List[Union[Handler, str]]) – A list of Handler objects or the string names for the photons that need to be reloaded.
Returns:: A list of updated Handler objects after the atomic reload process.
Return type:: List[Handler]

:raises Any exception that may occur during the _reload_photon(photon)() call.:

Notes

This function reloads a list of photon handlers or entire photons atomically and returns the updated Handler objects. The function loops through all provided handlers, atomically updates each of them using the previously defined _reload_photon(photon)() function, and collects the updated Handler objects in a list. If the result of the _reload_photon(photon)() function call returns a list of Handler objects, this function concatenates them onto the handlers list. If the result of the _reload_photon(photon)() function call returns a single Handler object, this function appends the single Handler object onto the handlers list.

Please note that any exception that is raised during the _reload_photon(photon)() function call is immediately propagated back to the caller. The exceptions may arise due to coding bugs, configuration issues, or other environmental reasons.

async _observe_photons(photons_directory, loop_trace=None)

|coro|

Observes photons for changes by continuously calculating their checksum and reloading when a change is detected.

Parameters:

photons_directory (str) – The path to the photons directory which needs to be monitored.
loop_trace (LoopTrace) – An optional trace object to keep track of loop iterations and allow testing of infinite loops, defaults to None.

Returns:

The function returns True when the observer loop has been terminated.

Return type:

bool

Raises:

Any exception that may occur during: –

self.load_photons(photons_directory)()
self.reload_photons(photons_to_reload)()
loop_trace.evalutate_tasks()
SystemUtils.continue_async(1)()

Notes

This function observes photons for changes by consistently calculating their checksum and reloading when a change is detected. The function loops indefinitely until the _is_watching flag is set to False. The observation process starts by calling the load_photons(photons_directory)() function to collect all the available photons. The function then makes a copy of all the _photons, and then loops through this copy and calculates the current checksum of each photon.

If the current photon’s checksum is different from the previous checksum value, the photon’s filepath is appended to the photons_to_reload list.
Once all the photons have been observed, the function checks whether photons_to_reload has any new entries. If new entries exist, the function calls the reload_photons(photons_to_reload)() function to atomically update the observed photons.
The function then pauses for one second by calling the SystemUtils.continue_async(1)() function before resuming with the next cycle. In case the loop_trace parameter is provided, this function evaluates the provided loop tasks by calling the loop_trace.evalutate_tasks() function.
Please note that any exception that is raised during any defined calls in this function is propagated back to the caller. The exceptions may arise due to coding bugs, configuration issues, or other environmental reasons.

async load_photon(photon_path, photon_base=IPhoton, other_classes=[], recursive=False)

|coro|

Loads a photon module from a given path, and initializes a corresponding Handler object. The provided path must be an existing file path to a photon module, and not a directory. This function takes an optional base class for the photon module, and a list of other class names to be loaded. If the recursive flag is set to True, this method loads all photon modules recursively from the path. The method raises a PhotonNotFoundError error if the provided path is not found, or if it is a directory.

Parameters:

photon_path (str) – A string representing the path to the photon module file to be loaded.
photon_base (type) – An optional base class type for photon modules to be loaded. Defaults to IPhoton.
other_classes (list[str]) – An optional list of string names of other classes to load from the photon module. Defaults to [].
recursive (bool) – A bool flag that indicates whether to search for photon modules recursively under the provided path. Defaults to False.

Returns:

A list of Handler objects representing the loaded photons, or an empty list if no photons are loaded successfully.

Return type:

bool

Raises:

PhotonNotFoundError – An error if the provided photon path doesn’t exist or is a directory.

Examples

>>> photon = await load_photon("photons/my_photon.py")
... # A photon will then be emitted and ready to start.

async load_photons(photons_directory, photon_base=IPhoton, other_classes=[], recursive=False)

|coro|

Loads a directory of photon modules, each module being represented by a corresponding Handler object, and returns a list of Handler objects. The provided path must be an existing directory path, and not a file. This function takes an optional base class for photon modules, and a list of other class names to be loaded. If the recursive flag is set to True, all photon modules will be loaded recursively from the directory.

Parameters:

photons_directory (str) – Represents the path to the directory containing the photon modules to be loaded.
photon_base (Optional[type]) – An optional base class type for photon modules to be loaded. Defaults to IPhoton.
other_classes (Optional[list[str]]) – An optional list of string names of other classes to load from the photon. Defaults to [].
recursive (Optional[bool]) – A flag that indicates whether to search for photon modules recursively under the provided directory path. Defaults to False.

Returns:

A collection of objects representing the loaded photons, or an empty list if no photons were loaded successfully.

Return type:

list[Handler]

Raises:

DirectoryNotFoundError – If the provided photon path doesn’t exist, or is a file.

Examples

>>> photons = await load_photons("photons/")
... # A list of photons will then be emitted and ready to start.

async unload_photon(photon, force_stop=False)

|coro|

Stops and unloads a specified photon from the loader’s photon registry by calling its finalizer, if any, and disposing unused resources. If the given photon is a string, it is looked up in the registry by filepath. The method returns a boolean value indicating whether the photon was successfully stopped and unloaded.

Important

Unloading a single photon unloads the entire file from which that photon originated. So, that means all other photons that may be found within the file as well.

Parameters:

photon (Handler | str) – The photon handler object or filepath which represents a running photon to be unloaded.
force_stop (bool) – A flag which allows a photon to be immediately halted and collected without calling its finalizer.

Returns:

A flag indicating if the provided photon was unloaded.

Return type:

bool

Examples

>>> await unload_photon(handler)
True
>>> await unload_photon('photons/a.py', force_stop=True)
True

async unload_photons(photons, force_stop=False)

|coro|

Stops and unloads a collection of photons from the loader’s photon registry by calling their finalizers, if any, and disposing unused resources. If a given photon is a string, it is looked up in the registry by filepath. The function returns a list of filepaths indicating whether the photons were successfully stopped and unloaded.

Important

Unloading any photon unloads the entire file from which that photon originated. So, that means all other photons that may be found within the file as well.

Parameters:

photons (list[Handler|str]) – A list of photon handler objects or filepaths to be unloaded.
force_stop (bool) – A flag which allows a photon to be immediately halted and collected without calling its finalizer.

Returns:

A list containing the filepaths of all unloaded photons.

Return type:

list[str]

Examples

>>> await unload_photons([handler1, handler2])
['photons/a.py', 'photons/a.py']
>>> await unload_photons(['photons/a.py', 'photons/b.py'], force_stop=True)
['photons/a.py', 'photons/b.py']

async reload_photon(photon)

|coro|

Atomically reloads the specified photon and provides a fresh handler.

Parameters:: photon (Handler|str) – The photon handler or filepath of a photon that is to be reloaded.
Returns:: A handler or list of handler objects representing the newly updated photon.
Return type:: Handler|list[Handler]

Notes

This function makes a copy of both the loader’s self._photons, and the related sys.modules. After which the provided photon will be unloaded, loaded again, and then returned as a freshly updated photon Handler object. If any errors, in either unloading or loading, all photons will revert to their original states.

Examples

>>> reloaded_handler = await reload_photon(my_photon_handler)
... # A new handler will be created and returned from the original handler object.
>>> reloaded_handler = await reload_photon('photons/my_photon.py')
... # A new handler will be created and returned from the provided handler file path.

async reload_photons(photons)

|coro|

Atomically reloads the specified photons and provides a list of fresh handlers.

Returns:: A list of handler objects representing the newly updated photons.
Return type:: list[Handler]
Parameters:: photons (list[luminal.managers.handler.Handler | str]) –

Notes

This function makes a copy of both the loader’s self._photons, and the related sys.modules. After which the provided photons will be unloaded, loaded again, and then returned as freshly updated photon Handler objects. If any errors, in either unloading or loading, all photons will revert to their original states.

Examples

>>> reloaded_handlers = await reload_photons([my_photon_handler, second_photon_handler])
... # New handlers will be created and returned from the original handler objects.
>>> reloaded_handlers = await reload_photons('photons/my_photon.py', 'photons/second_photon.py')
... # New handlers will be created and returned from the provided handler file path.

async watch_photons(photons_directory)

|coro|

Creates a thread manager and spawns a new thread to monitor photons in a given directory for changes and reloads them atomically.

Parameters:

photons_directory (str) – The directory path to the photons folder which needs to be monitored.

Raises:

TypeError – If the photons_directory argument is not a string.
DirectoryNotFoundError – If the photons_directory argument is not a valid directory.
ThreadManagerAlreadyRunning – If the loader is already monitoring photons.

Return type:

None

Notes

This asynchronous function spawns a thread manager and a new thread to monitor photons in a given directory.
The function starts by validating whether the photons_directory argument is a valid directory and of type str.
If the _is_watching flag is True, the function raises a warning by raising ThreadManagerAlreadyRunningError.
Then the function sets the _is_watching flag to True and sets the number of allowed threads to one. The function appends a new thread to the thread manager through the _start_watching() callback function. The thread is then started and runs indefinitely until the _is_watching flag is set to False. The wait time between cycles is set to one second by calling the SystemUtils.continue_async(1)() function.
Please note that this function starts the thread manager and a new thread to monitor photons. Hence, the thread manager must be stopped by calling the stop_watching_photons() function of the photon Loader instance.

async stop_watching_photons(halt_threads=False)

|coro|

Stops all threads from watching changes in photons and gracefully stops the thread manager.

Parameters:: halt_threads (Optional[bool]) – An optional boolean flag to halt all running threads immediately without a flag, instead of gracefully stopping the thread manager, defaults to False.
Return type:: None

Notes

This asynchronous function stops all threads and gracefully shuts down the thread manager from watching changes in photons.

The function starts by setting the _is_watching flag to False. If halt_threads is True, the function calls the halt() function of the thread manager instance to stop all running threads immediately. If halt_threads is False, the function calls the stop() function to gracefully stop all threads.
To prevent memory leaks and other issues, it is recommended to stop any running threads gracefully instead of abruptly halting or killing them. In most cases, setting halt_threads to False is preferred to ensure a graceful shutdown of all threads.

luminal.managers.loader

Module Contents

Classes

`luminal.managers.loader`