koji_habitude.workflow

koji_habitude.workflow

Workflow class for orchestrating the synchronization process.

The steps and imports required for working with habitude are long and involved, so this class provides a simple interface for orchestrating the workflow, allowing for users to focus on the data and configuration, rather than the implementation details. The interface is designed to be extensible, allowing for users to override the default behavior for each step of the workflow. The workflow process is also designed to be pauseable, allowing for users to pause the workflow and resume it later.

author:: Christopher O’Brien <obriencj@gmail.com>
license:: GNU General Public License v3
ai-assistant:: Claude 3.5 Sonnet via Cursor

class koji_habitude.workflow.ApplyDictWorkflow(objects: List[Dict[str, Any]], template_paths: List[str | Path] | None = None, profile: str = 'koji', chunk_size: int = 100, skip_phantoms: bool = False)

Bases: DictWorkflow

Workflow for applying data onto a Koji hub instance, using pre-created dictionaries of objects.

This is used by the template apply command to apply a single template’s expansion onto a Koji hub instance.

class koji_habitude.workflow.ApplyWorkflow(paths: List[str | Path], template_paths: List[str | Path] | None = None, recursive: bool = False, profile: str = 'koji', chunk_size: int = 100, skip_phantoms: bool = False)

Bases: Workflow

Workflow for applying data onto a Koji hub instance. Implements the majority of the behavior of the apply command.

class koji_habitude.workflow.CompareDictWorkflow(objects: List[Dict[str, Any]], template_paths: List[str | Path] | None = None, profile: str = 'koji', chunk_size: int = 100, skip_phantoms: bool = False)

Bases: DictWorkflow

Workflow for comparing data against a Koji hub instance, using pre-created dictionaries of objects.

This is used by the template compare command to compare a single template’s expansion against the objects on a Koji hub instance.

get_session(profile: str = 'koji') → ClientSession: Override the default session creation to not authenticate.

review_resolver_report(): Diff mode is allowed to have phantoms, so we don’t need to do anything.

class koji_habitude.workflow.CompareWorkflow(paths: List[str | Path], template_paths: List[str | Path] | None = None, recursive: bool = False, profile: str = 'koji', chunk_size: int = 100, skip_phantoms: bool = False)

Bases: Workflow

Workflow for comparing data against a Koji hub instance. Implements the majority of the behavior of the compare command. Similar to the ApplyWorkflow in most aspects, but with a processor that omits the apply operations.

get_session(profile: str = 'koji') → ClientSession: Override the default session creation to not authenticate.

review_resolver_report(): Diff mode is allowed to have phantoms, so we don’t need to do anything.

class koji_habitude.workflow.DictWorkflow(paths: ~typing.List[str | ~pathlib.Path] | None = None, template_paths: ~typing.List[str | ~pathlib.Path] | None = None, recursive: bool = False, profile: str = 'koji', chunk_size: int = 100, skip_phantoms: bool = False, cls_multiloader: ~typing.Type[~koji_habitude.loader.MultiLoader] = <class 'koji_habitude.loader.MultiLoader'>, cls_yamlloader: ~typing.Type[~koji_habitude.loader.YAMLLoader] = <class 'koji_habitude.loader.YAMLLoader'>, cls_template_namespace: ~typing.Type[~koji_habitude.namespace.TemplateNamespace] = <class 'koji_habitude.namespace.TemplateNamespace'>, cls_namespace: ~typing.Type[~koji_habitude.namespace.Namespace] = <class 'koji_habitude.namespace.Namespace'>, cls_processor: ~typing.Type[~koji_habitude.processor.Processor] = <class 'koji_habitude.processor.Processor'>, cls_resolver: ~typing.Type[~koji_habitude.resolver.Resolver] = <class 'koji_habitude.resolver.Resolver'>, cls_solver: ~typing.Type[~koji_habitude.solver.Solver] = <class 'koji_habitude.solver.Solver'>, objects: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>)

Bases: Workflow

Workflow for operating over pre-created dictionaries of objects, rather than using a loader to pull data from a filesystem.

load_data(paths: List[str | Path], templates: TemplateNamespace | None = None) → Namespace

objects: List[Dict[str, Any]]

class koji_habitude.workflow.Workflow(paths: ~typing.List[str | ~pathlib.Path] | None = None, template_paths: ~typing.List[str | ~pathlib.Path] | None = None, recursive: bool = False, profile: str = 'koji', chunk_size: int = 100, skip_phantoms: bool = False, cls_multiloader: ~typing.Type[~koji_habitude.loader.MultiLoader] = <class 'koji_habitude.loader.MultiLoader'>, cls_yamlloader: ~typing.Type[~koji_habitude.loader.YAMLLoader] = <class 'koji_habitude.loader.YAMLLoader'>, cls_template_namespace: ~typing.Type[~koji_habitude.namespace.TemplateNamespace] = <class 'koji_habitude.namespace.TemplateNamespace'>, cls_namespace: ~typing.Type[~koji_habitude.namespace.Namespace] = <class 'koji_habitude.namespace.Namespace'>, cls_processor: ~typing.Type[~koji_habitude.processor.Processor] = <class 'koji_habitude.processor.Processor'>, cls_resolver: ~typing.Type[~koji_habitude.resolver.Resolver] = <class 'koji_habitude.resolver.Resolver'>, cls_solver: ~typing.Type[~koji_habitude.solver.Solver] = <class 'koji_habitude.solver.Solver'>)

Bases: object

A Workflow represents the combination of the individual features and steps of koji-habitude, providing a simplified interface for loading YAML, feeding and validating the data into a Namespace, expanding templates, resolving dependencies, identifying change sets, and applying the relevant changes onto a Koji hub instance.

chunk_size: int = 100

cls_multiloader: alias of MultiLoader

cls_namespace: alias of Namespace

cls_processor: alias of Processor

cls_resolver: alias of Resolver

cls_solver: alias of Solver

cls_template_namespace: alias of TemplateNamespace

cls_yamlloader: alias of YAMLLoader

dataseries: List[CoreObject] = None

get_session(profile: str = 'koji') → ClientSession

iter_run()

load_data(paths: List[str | Path], templates: TemplateNamespace | None = None) → Namespace

load_templates(paths: List[str | Path]) → TemplateNamespace

load_yaml(paths: List[str | Path]) → Iterator[Dict[str, Any]]

namespace: Namespace = None

paths: List[str | Path] = None

processor: Processor = None

processor_step_callback(step: int, handled: int): Callback for the processor, invoked after each processor.step() invocation.

profile: str = 'koji'

recursive: bool = False

resolver: Resolver = None

resolver_report: ResolverReport = None

resume(): Resume a paused workflow, starting from the current state and iterating over the phases. As the workflow progresses, state transitions are triggered, and the overridable callback workflow_state_change is invoked. If the callback returns True, the workflow is paused and this method returns True. If the workflow completes successfully, this method returns False.

review_resolver_report(): Review the resolver report and raise an exception if there are any phantoms.

run() → WorkflowState

Run the workflow, starting from the READY state and iterating over the phases. As the workflow progresses, state transitions are triggered, and the overridable callback workflow_state_change is invoked. If the callback returns True, the workflow is paused and this method returns the current state. If the workflow completes successfully, this method returns WorkflowState.COMPLETED.

A paused workflow can be resumed by calling the resume method, which will pick up where the workflow left off, and may be paused again.

If an exception is raised during the run, the workflow state is set to WorkflowState.FAILED and the exception is re-raised.

run_connecting()

run_loading()

run_processing()

run_resolving()

run_solving()

session: ClientSession = None

skip_phantoms: bool = False

solver: Solver = None

state: WorkflowState = 'ready'

state_change(from_state: WorkflowState, to_state: WorkflowState)

summary: ProcessorSummary = None

template_paths: List[str | Path] = None

workflow_paused(): Callback for the workflow, invoked when the workflow is paused.

workflow_state_change(from_state: WorkflowState, to_state: WorkflowState) → bool: Callback for the workflow, invoked during the phases of the Workflow.run() invocation.

exception koji_habitude.workflow.WorkflowPhantomsError(message: str, report: ResolverReport)

Bases: Exception

Exception raised when the workflow has phantoms.

class koji_habitude.workflow.WorkflowState(value)

Bases: Enum

The states of the workflow.

See the Workflow.run and Workflow.resume methods for more information.

These values are passed to the callback Workflow.workflow_state_change when the workflow transitions between states. The callback can return True to pause the workflow, in which case Workflow.resume can be called to resume the workflow from the current state.

COMPLETED = 'completed'

CONNECTED = 'connected'

CONNECTING = 'connecting'

FAILED = 'failed'

LOADED = 'loaded'

LOADING = 'loading'

PROCESSED = 'processed'

PROCESSING = 'processing'

READY = 'ready'

RESOLVED = 'resolved'

RESOLVING = 'resolving'

SOLVED = 'solved'

SOLVING = 'solving'

STARTING = 'starting'

exception koji_habitude.workflow.WorkflowStateError

Bases: Exception

Exception raised when the workflow is in an invalid state.