orchestrata_ultimate.refactor.codemod.observability.events

Reference › orchestrata_ultimate › refactor › codemod › observability › events

The event vocabulary of the codemod engine, on top of the observer pattern in core.patterns.observer. CodemodEventType enumerates the lifecycle events: workspace loaded, index built, pass started and completed, file mutated, review started and completed, finding emitted, commit started, completed and aborted, rollback triggered, and error. A CodemodEvent carries a type, a timestamp, a data dictionary and a correlation id that ties together all events of one operation. The workspace and the transaction layer emit them, report and the five observers consume or reference the types, and the facade re-exports them.

At a glance

Class diagram

Dependency diagram

API

Classes

CodemodEventType (enum)

Bases: Enum · Decorators: traces(spec='110', story='US-110-005', reqs=['REQ-110-025']) · Line 14

The 13 lifecycle event kinds REQ-110-025 names.

CodemodEvent

Line 37

Codemod event.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/observability/events.py

"""SPEC-110 codemod engine module."""

from __future__ import annotations

from enum import Enum

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.patterns.observer import Observer

__all__ = ["CodemodEvent", "CodemodEventType", "Observer"]


@traces(spec="110", story="US-110-005", reqs=["REQ-110-025"])
class CodemodEventType(Enum):
    """The 13 lifecycle event kinds REQ-110-025 names.

    ``ERROR`` is kept as an alias of ``ERROR_DETECTED`` (same value) so the
    observers and callers written against the original name keep working.
    """

    WORKSPACE_LOADED = "workspace_loaded"
    INDEX_BUILT = "index_built"
    PASS_STARTED = "pass_started"
    PASS_COMPLETED = "pass_completed"
    FILE_MUTATED = "file_mutated"
    REVIEW_STARTED = "review_started"
    REVIEW_COMPLETED = "review_completed"
    FINDING_EMITTED = "finding_emitted"
    COMMIT_STARTED = "commit_started"
    COMMIT_COMPLETED = "commit_completed"
    COMMIT_ABORTED = "commit_aborted"
    ROLLBACK_TRIGGERED = "rollback_triggered"
    ERROR_DETECTED = "error"
    ERROR = "error"


class CodemodEvent:
    """Codemod event."""

    def __init__(
        self, type: CodemodEventType, timestamp: float, data: dict, correlation_id: str
    ) -> None:
        """Wire CodemodEvent's composed collaborators.

        Args:
            type: Field value for ``type``.
            timestamp: Field value for ``timestamp``.
            data: Field value for ``data``.
            correlation_id: Field value for ``correlation_id``.
        """
        self.type = type
        self.timestamp = timestamp
        self.data = data
        self.correlation_id = correlation_id