gfw.common.config#

Defines the PipelineConfig class used to configure data pipeline executions.

It includes: - A dataclass PipelineConfig that stores date ranges and any unknown arguments. - A custom exception PipelineConfigError for handling invalid configuration inputs.

Intended for use in CLI-based or programmatic pipeline setups where date ranges and additional arguments need to be passed and validated.

Classes

PipelineConfig

Configuration object for data pipeline execution.

Exceptions

PipelineConfigError

Custom exception for pipeline configuration errors.

exception PipelineConfigError[source]#

Custom exception for pipeline configuration errors.

add_note()#: Exception.add_note(note) – add a note to the exception

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class PipelineConfig(*, date_range, name='', version='0.1.0', jinja_folder='assets/queries', mock_bq_clients=False, unknown_parsed_args=<factory>, unknown_unparsed_args=())[source]#

Configuration object for data pipeline execution.

Note

This class is completely generic and independent of any specific pipeline framework.

date_range: tuple[str, str]#: Tuple of start and end dates in ISO format (YYYY-MM-DD).

name: str = ''#: Name of the pipeline.

version: str = '0.1.0'#: Version of the pipeline.

jinja_folder: str = 'assets/queries'#: The folder that contains the jinja2 templates.

mock_bq_clients: bool = False#: If True, all BigQuery interactions will be mocked.

unknown_parsed_args: dict[str, Any]#: Parsed CLI or config arguments not explicitly defined in self.

unknown_unparsed_args: tuple[str, ...] = ()#: Raw unparsed CLI arguments.

classmethod from_namespace(ns, **kwargs)[source]#

Creates a PipelineConfig instance from a types.SimpleNamespace.

Parameters:

ns (SimpleNamespace) – Namespace containing attributes matching this PipelineConfig fields.
**kwargs (Any) – Any additional arguments to be passed to the class constructor.

Returns:

A new PipelineConfig instance.

Return type:

PipelineConfig

property parsed_date_range: tuple[date, date]#

Returns the parsed start and end dates as date objects.

Raises:: PipelineConfigError – If any of the dates are not in valid ISO format.
Returns:: A tuple containing parsed start and end dates.

property top_level_package: str#: Returns the top-level package from this module.

property jinja_env: Environment#: Returns a default jinja2 environment.

property start_date: date#

Returns the start date of the configured range.

Returns:: A date object representing the start of the range.

property end_date: date#

Returns the end date of the configured range.

Returns:: A date object representing the end of the range.

property pre_hooks: Sequence[Callable[[Any], None]]#: Sequence of callables executed before pipeline run.

property post_hooks: Sequence[Callable[[Any], None]]#: Sequence of callables executed after successful pipeline run.

to_dict()[source]#

Converts a PipelineConfig instance to dictionary.

Returns:: A dictionary representation of the configuration.
Return type:: dict[str, Any]