fluxnet_shuttle.core.registry

Plugin Registry and Error Collection

module::

fluxnet_shuttle.core.registry

synopsis:

Plugin registry and error collection for FLUXNET Shuttle Library

moduleauthor:

Valerie Hendrix <vchendrix@lbl.gov>

moduleauthor:

Sy-Toan Ngo <sytoanngo@lbl.gov>

platform:

Unix, Windows

created:

2025-10-09

updated:

2025-12-09

This module provides the plugin registry for managing data hub plugins and error collection capabilities.

Functions

dataclass([cls, init, repr, eq, order, ...])	Add dunder methods based on the fields defined in the class.
field(*[, default, default_factory, init, ...])	Return an object to identify dataclass fields.

Classes

Any(*args, **kwargs)	Special type indicating an unconstrained type.
DataHubPlugin([config])	Base class for all data hub plugins.
ErrorCollectingIterator(plugins, operation, ...)	Async iterator that collects errors while yielding results.
ErrorSummary(**data)	Pydantic model for error summary information.
FluxnetDatasetMetadata(**data)	Combined model for complete FLUXNET dataset metadata.
PluginErrorDetail(**data)	Pydantic model for individual plugin error details.
PluginErrorInfo(plugin_name, error[, ...])	Container for plugin execution errors.
PluginRegistry()	Registry for managing data hub plugins with automatic discovery.
datetime(year, month, day[, hour[, minute[, ...)	The year, month and day arguments are required.

class fluxnet_shuttle.core.registry.ErrorCollectingIterator(plugins, operation, **kwargs)

Bases: object

Async iterator that collects errors while yielding results.

This class implements the async iterator protocol and collects results from multiple plugins while isolating and collecting any errors that occur.

Parameters:

• plugins (Dict[str, DataHubPlugin])

• operation (str)

• kwargs (Any)

add_error(plugin_name, error, operation='')

Add an error to the collection.

Parameters:

• plugin_name (str) – Name of the plugin that encountered the error

• error (Exception) – The exception that occurred

• operation (str) – The operation being performed when the error occurred

Return type:

None

get_error_summary()

Get summary of all errors.

Returns:

Pydantic model containing error summary information

Return type:

ErrorSummary

has_errors()

Check if any errors were collected.

Returns:

True if any errors occurred

Return type:

bool

class fluxnet_shuttle.core.registry.PluginErrorInfo(plugin_name, error, timestamp=<factory>, operation='')

Bases: object

Container for plugin execution errors.

Parameters:

• plugin_name (str)

• error (Exception)

• timestamp (datetime)

• operation (str)

error: Exception

operation: str = ''

plugin_name: str

timestamp: datetime

class fluxnet_shuttle.core.registry.PluginRegistry

Bases: object

Registry for managing data hub plugins with automatic discovery.

This class manages the registration and instantiation of data hub plugins, including automatic discovery through entry points.

create_instance(name, **config)

Create an instance of a plugin.

Parameters:

• name (str) – Plugin name

• **config (Any) – Configuration parameters

Return type:

DataHubPlugin

Returns:

Plugin instance

Raises:

ValueError – If plugin is not found

get_plugin(name)

Get a plugin by name.

Parameters:

name (str) – Plugin name

Return type:

Type[DataHubPlugin]

Returns:

Plugin class or None if not found

Raises:

ValueError – If plugin is not found

list_plugins()

List all registered plugin names.

Return type:

List[str]

Returns:

List of plugin names

register(plugin_class)

Register a data hub plugin.

Parameters:

plugin_class (Type[DataHubPlugin]) – The plugin class to register

Return type:

None