koji_habitude.processor

koji_habitude.processor

Core processing engine for synchronizing koji objects with a hub instance. Handles the read/compare/write cycle in chunks with multicall optimization.

author:

Christopher O’Brien <obriencj@gmail.com>

license:

GNU General Public License v3

ai-assistant:

Claude 3.5 Sonnet via Cursor

class koji_habitude.processor.CompareOnlyProcessor(koji_session: ClientSession, dataseries: Iterable[CoreObject], resolver: Resolver, chunk_size: int = 100, skip_phantoms: bool = False, spinner_fn: Callable[[str], None] | None = None)

Bases: Processor

Processor variant that only performs read and compare operations.

Useful for the ‘diff’ command to show what changes would be made without actually applying them to the koji instance.

step_apply() → None

Override to skip write operations in diff mode.

Logs what changes would be applied instead of actually applying them.

class koji_habitude.processor.Processor(koji_session: ClientSession, dataseries: Iterable[CoreObject], resolver: Resolver, chunk_size: int = 100, skip_phantoms: bool = False, spinner_fn: Callable[[str], None] | None = None)

Bases: object

Processes a stream of koji objects in dependency-resolved order.

Executes the read/compare/write cycle in chunks, using koji multicalls for efficient batch operations. Each object in the stream is expected to have methods for fetching state, comparing with koji data, and applying changes.

is_broken() → bool

Check if the processor is in the BROKEN state.

Returns:

True if the processor is in the BROKEN state, False otherwise

is_exhausted() → bool

Check if the object stream has been fully processed.

Returns:

True if the processor is in the EXHAUSTED state, False otherwise

run(step_callback: Callable[[int, int], None] | None = None) → ProcessorSummary

Process all objects in the stream by repeatedly calling step() until we are in an EXHAUSTED or BROKEN state. Returns a summary of the processing results.

Returns:

Summary of processing results including total objects processed, changes applied, and any errors encountered.

spin(message: str | None = None) → None

Call the spinner function with an optional message. The spinner function can tick over whether or not there’s a new message.

step(chunk_size: int | None = None) → int

Execute one complete cycle: read -> compare -> apply.

Can be safely invoked from either the READY_CHUNK or READY_READ states. If READY_CHUNK, the current chunk is discarded and a new one is loaded. If there was no chunk, state is set to EXHAUSTED and 0 is returned. Otherwise, state is set to READY_READ and step_read, step_compare, and step_apply are called in order. If any of these steps fail, the state is set to BROKEN and an exception is raised.

Returns:

count of objects processed

step_apply() → None

Apply the identified changes to the koji instance.

This step:

1. Retrieves the change reports with changes identified in step_compare()

2. Calls apply() on each report to execute the changes via multicall

3. Commits all changes to the koji instance

After this step, the koji instance matches the desired state.

step_compare() → None

Compare each object with its current koji state and identify changes.

This step:

1. Retrieves the change reports populated in step_read()

2. Calls compare() on each report to analyze differences

3. Populates the reports with specific changes that need to be made

After this step, change reports contain both current data and required changes.

step_read() → None

Fetch current state from Koji for all objects in current chunk.

This step:

1. Creates empty change reports for each object via obj.change_report()

2. Calls read() on each report to fetch current koji state via multicall

3. Stores the populated reports for use in step_compare()

After this step, change reports contain current koji data but no changes yet.

class koji_habitude.processor.ProcessorState(value)

Bases: Enum

Processor state machine for managing the read/compare/apply cycle.

State transitions:

READY_CHUNK → READY_READ → READY_COMPARE → READY_APPLY → READY_CHUNK

Any state → BROKEN (on error)

READY_CHUNK → EXHAUSTED (when stream empty)

BROKEN = 'broken'

EXHAUSTED = 'exhausted'

READY_APPLY = 'ready-apply'

READY_CHUNK = 'ready-chunk'

READY_COMPARE = 'ready-compare'

READY_READ = 'ready-read'

exception koji_habitude.processor.ProcessorStateError

Bases: Exception

class koji_habitude.processor.ProcessorSummary(total_objects: int, steps_completed: int, state: koji_habitude.processor.ProcessorState, change_reports: Dict[Tuple[str, str], koji_habitude.models.change.ChangeReport])

Bases: object

change_reports: Dict[Tuple[str, str], ChangeReport]

state: ProcessorState

steps_completed: int

property total_changes: int

total_objects: int