pipeline.nameSpace

Classes

class pipeline.nameSpace.LocalFunctionProvider

Abstract base class for local function providers.

This class serves as a blueprint for creating custom local function providers. Subclasses must implement the get_function method to generate appropriate function objects based on the context.

Examples

.. code-block:: python

   class MyCustomProvider(LocalFunctionProvider):
       def get_function(self) -> Callable:
           filter_by = self.context.get('filter_by')

           if filter_by == 'status':

Inherits from:: ABC

Methods:

__init__(): Initialize with optional context.

update_context(context: dict[str, Any]) → None: Update the context dictionary with new values.

run_function() → Any

Run the function based on the context.

This method must be implemented by subclasses to generate an appropriate function object based on the current context.

Returns

Any

Any
    The result of the function execution

Functions

pipeline.nameSpace.register_module(module_class: Type[PipeModule], silent: bool = False) → None

Register a user-defined PipeModule class.

Parameters

module_classThe PipeModule subclass to register: silent: If True, suppresses printed messages (default: False)

Returns

Any: None

Examples

.. code-block:: python

       # In your custom module file (e.g., my_modules/custom_module.py)
       from gdi.modules.pipeline import PipeModule, status_manage

       @status_manage
       class MyCustomModule(PipeModule):
           # Your module implementation
           pass

       # In your main file where you want to register it
       from gdi.modules.nameSpace import register_module
       from my_modules.custom_module import MyCustomModule

       # Register your custom module (with or without messages)
       register_module(MyCustomModule)  # With messages
       register_module(MyCustomModule, silent=True)  # Without messages

pipeline.nameSpace.register_modules_from_file(file_path: str | Path, silent: bool = False) → None

Register all PipeModule subclasses from a specific Python file.

Parameters

file_pathPath to the Python file containing module definitions: silent: If True, suppresses printed messages (default: False)

Returns

Any: None

Examples

.. code-block:: python

       from gdi.modules.nameSpace import register_modules_from_file

       # With messages
       register_modules_from_file("./my_modules/custom_module.py")

       # Without messages
       register_modules_from_file("./my_modules/custom_module.py", silent=True)

pipeline.nameSpace.register_modules_from_directory(directory_path: str | Path, silent: bool = False) → None

Scan a directory and register all PipeModule subclasses found.

Parameters

directory_pathPath to the directory containing module definitions: silent: If True, suppresses printed messages (default: False)

Returns

Any: None

Examples

.. code-block:: python

       from gdi.modules.nameSpace import register_modules_from_directory

       # With messages
       register_modules_from_directory("./usermodules")

       # Without messages
       register_modules_from_directory("./usermodules", silent=True)

pipeline.nameSpace.discover_modules_in_directory(package_name: str, silent: bool = False) → dict[str, Type[PipeModule]]

Dynamically discover all PipeModule subclasses.

Parameters

package_nameThe name of the package to scan (‘gdi.modules’ or ‘gdisdk.modules’): silent: If True, suppresses printed messages (default: False)

Returns

Any: A dictionary mapping class names to class objects

pipeline.nameSpace.get_module_namespace(silent: bool = False) → dict[str, Type[PipeModule]]

Get all PipeModule subclasses from the gdi.modules or gdisdk.modules package,

flow controllers, and user registry.

Returns

Any: A dictionary mapping class names to class objects

pipeline.nameSpace.local_function(func: Callable) → Callable

Decorator to mark a function as a local function for PythonCoder module.

This decorator provides a simpler alternative to creating a class that inherits from LocalFunctionProvider. The decorated function receives the module instance as its first argument, providing access to all ports and attributes.

Parameters

funcCallable: The function to decorate. Must accept module as its first parameter. For UI schema functions, can also accept reset as a second parameter.

Returns

“param1”: UIAttributeSchema(…) }

Examples

.. code-block:: python

   from gdi.pipeline.nameSpace import local_function

   @local_function
   def process_tables(module):
       '''Main execution function'''
       tables = module["InputTables"]
       # ... processing logic ...

Notes

- The function must accept ``module`` as its first parameter
- For execution functions: return ``{port_name: data}`` dict
- For UI schema functions: return ``{attr_name: UIAttributeSchema}`` dict
- Functions are registered globally and can be looked up by name
- When using in the same script (no serialization), pass the function directly
  to PythonCoder's ``local_function`` parameter

pipeline.nameSpace.get_local_function_from_registry(function_name: str) → Callable | None

Get a local function from the global registry.

Parameters

function_namestr: The name of the function to retrieve.

Returns

Any

Callable | None
    The registered function, or None if not found.

pipeline.nameSpace.load_local_function(file_path: str | Path, function_name: str, debug_mode: bool = False) → Callable | LocalFunctionProvider

Load a local function from a Python file.

This function dynamically loads either:

A function decorated with @local_function (new approach, recommended)
A LocalFunctionProvider subclass (legacy approach)

Parameters

file_pathstr | Path: The path to the Python file containing the local function or provider class.
function_namestr: The name of the function or LocalFunctionProvider subclass to load. debug_mode : bool, default False If True, provides detailed error messages including full tracebacks. If False, provides concise error messages for production use.

Returns

Any

Callable | LocalFunctionProvider
    Either a callable function (new approach) or LocalFunctionProvider instance (legacy)

Examples

.. code-block:: python

   # Load a decorated function (new approach)
   func = load_local_function('my_functions.py', 'process_tables')
   result = func(module)  # Call directly with module

   # Load a LocalFunctionProvider (legacy approach)
   provider = load_local_function('my_providers.py', 'MyProvider')
   provider.update_context(context)
   result = provider.run_function()

pipeline.nameSpace.load_local_function_provider(file_path: str | Path, class_name: str, debug_mode: bool = False) → Callable | LocalFunctionProvider: Deprecated: Use load_local_function instead.

pipeline.nameSpace.get_local_function(module: PipeModule, local_functions_path: str | Path, function_name: str, debug_mode: bool = False) → LocalFunctionProvider | None

Get the local function of the module.

This function is primarily for legacy LocalFunctionProvider support. For new code, prefer using load_local_function() directly and calling the function with the module as the first argument.

Parameters

modulePipeModule: The module to get the local function.
local_functions_pathstr | Path: The path to the local functions file.
function_namestr: The name of the function to get. debug_mode : bool, default False If True, provides detailed error messages including full tracebacks when loading fails. If False, provides concise error messages for production use.

Notes

This function is kept for backward compatibility with AttributeCaculator.
New code should use load_local_function() and call the function directly.