Skip to content

events

Event types and reader for Snakemake logger plugin integration.

This module provides types and utilities for reading real-time events from the snakemake-logger-plugin-snakesee plugin. Events provide more accurate and timely job status information than log parsing.

Classes

EventReader

Streaming reader for snakesee event files.

Reads events incrementally from a JSONL file, tracking the current position to only return new events on subsequent calls.

Attributes:

Name Type Description
event_file

Path to the event file.
Source code in snakesee/events.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class EventReader:
    """Streaming reader for snakesee event files.

    Reads events incrementally from a JSONL file, tracking the current
    position to only return new events on subsequent calls.

    Attributes:
        event_file: Path to the event file.
    """

    def __init__(self, event_file: Path) -> None:
        """Initialize the event reader.

        Args:
            event_file: Path to the event file.
        """
        self.event_file = event_file
        self._offset: int = 0
        self._lock = threading.RLock()

    def read_new_events(self) -> list[SnakeseeEvent]:
        """Read events added since last call.

        Returns:
            List of new events. Empty list if no new events or file doesn't exist.
        """
        if not self.event_file.exists():
            return []

        # Get current offset under lock (minimize lock hold time)
        with self._lock:
            offset = self._offset

        # Perform file I/O without holding the lock to avoid blocking
        events: list[SnakeseeEvent] = []
        new_offset = offset
        try:
            with open(self.event_file, "r", encoding="utf-8") as f:
                f.seek(offset)
                for line in f:
                    line = line.strip()
                    if line:
                        try:
                            events.append(SnakeseeEvent.from_json(line))
                        except (orjson.JSONDecodeError, ValueError, KeyError, TypeError) as e:
                            # Skip malformed lines but log for debugging
                            logger.debug(
                                "Skipping malformed event line: %s... (%s)",
                                line[:50],
                                e,
                            )
                            continue
                new_offset = f.tell()
        except OSError as e:
            # File access error - log and return empty list
            logger.warning("Error reading event file %s: %s", self.event_file, e)
            return events

        # Update offset under lock
        with self._lock:
            self._offset = new_offset

        return events

    def reset(self) -> None:
        """Reset reader to start of file.

        Call this to re-read all events from the beginning.
        """
        with self._lock:
            self._offset = 0

    @property
    def has_events(self) -> bool:
        """Check if the event file exists and has content.

        Returns:
            True if event file exists and is non-empty.
        """
        if not self.event_file.exists():
            return False
        return self.event_file.stat().st_size > 0

Attributes

has_events property 
has_events: bool

Check if the event file exists and has content.

Returns:

Type Description
bool

True if event file exists and is non-empty.

Methods:

__init__ 
__init__(event_file: Path) -> None

Initialize the event reader.

Parameters:

Name Type Description Default
event_file Path

Path to the event file.

 required
Source code in snakesee/events.py 
168
169
170
171
172
173
174
175
176
def __init__(self, event_file: Path) -> None:
    """Initialize the event reader.

    Args:
        event_file: Path to the event file.
    """
    self.event_file = event_file
    self._offset: int = 0
    self._lock = threading.RLock()
read_new_events 
read_new_events() -> list[SnakeseeEvent]

Read events added since last call.

Returns:

Type Description
list[SnakeseeEvent]

List of new events. Empty list if no new events or file doesn't exist.
Source code in snakesee/events.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def read_new_events(self) -> list[SnakeseeEvent]:
    """Read events added since last call.

    Returns:
        List of new events. Empty list if no new events or file doesn't exist.
    """
    if not self.event_file.exists():
        return []

    # Get current offset under lock (minimize lock hold time)
    with self._lock:
        offset = self._offset

    # Perform file I/O without holding the lock to avoid blocking
    events: list[SnakeseeEvent] = []
    new_offset = offset
    try:
        with open(self.event_file, "r", encoding="utf-8") as f:
            f.seek(offset)
            for line in f:
                line = line.strip()
                if line:
                    try:
                        events.append(SnakeseeEvent.from_json(line))
                    except (orjson.JSONDecodeError, ValueError, KeyError, TypeError) as e:
                        # Skip malformed lines but log for debugging
                        logger.debug(
                            "Skipping malformed event line: %s... (%s)",
                            line[:50],
                            e,
                        )
                        continue
            new_offset = f.tell()
    except OSError as e:
        # File access error - log and return empty list
        logger.warning("Error reading event file %s: %s", self.event_file, e)
        return events

    # Update offset under lock
    with self._lock:
        self._offset = new_offset

    return events
reset 
reset() -> None

Reset reader to start of file.

Call this to re-read all events from the beginning.

Source code in snakesee/events.py 
222
223
224
225
226
227
228
def reset(self) -> None:
    """Reset reader to start of file.

    Call this to re-read all events from the beginning.
    """
    with self._lock:
        self._offset = 0

EventType

Bases: str, Enum

Event types from the snakesee logger plugin.

These mirror the event types defined in the logger plugin.

Source code in snakesee/events.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
class EventType(str, Enum):
    """Event types from the snakesee logger plugin.

    These mirror the event types defined in the logger plugin.
    """

    WORKFLOW_STARTED = "workflow_started"
    JOB_SUBMITTED = "job_submitted"
    JOB_QUEUED = "job_queued"
    JOB_STARTED = "job_started"
    JOB_FINISHED = "job_finished"
    JOB_ERROR = "job_error"
    PROGRESS = "progress"

SnakeseeEvent dataclass

A single event from the logger plugin.

This is a frozen dataclass to ensure events are immutable once parsed.

Attributes:

Name Type Description
event_type EventType

Type of the event.
timestamp float

Unix timestamp when the event occurred.
job_id int | None

Snakemake job ID (for job events).
rule_name str | None

Name of the rule (for job events).
wildcards tuple[tuple[str, str], ...] | None

Wildcard values for the job.
threads int | None

Number of threads allocated to the job.
resources tuple[tuple[str, Any], ...] | None

Resource requirements for the job.
input_files tuple[str, ...] | None

Tuple of input file paths.
output_files tuple[str, ...] | None

Tuple of output file paths.
duration float | None

Job duration in seconds (for finished/error events).
error_message str | None

Error message (for error events).
completed_jobs int | None

Number of completed jobs (for progress events).
total_jobs int | None

Total number of jobs (for progress events).
workflow_id str | None

Unique workflow identifier.
executor str | None

Remote executor identifier (e.g. "aws-batch"), if applicable.
external_jobid str | None

External executor job id/ARN, for remote jobs.
remote_status str | None

Raw backend status string (e.g. "RUNNING"), for remote jobs.
queued_at float | None

Epoch seconds the job entered the remote queue, if known.
started_at float | None

Epoch seconds the job began executing on a remote node, if known.
stopped_at float | None

Epoch seconds the job stopped executing remotely, if known.
attempt int | None

1-based attempt number for retried/preempted remote jobs.
exit_code int | None

Container/process exit code for a finished remote job.
status_reason str | None

Backend-provided reason string (e.g. failure cause).
queue str | None

The remote queue the job was routed to, if known.
log_stream str | None

Backend log stream identifier (e.g. CloudWatch stream).
region str | None

Cloud region, used to build console deep links.
termination_category str | None

Why the job died (e.g. "spot", "oom"), if classified.
termination_source str | None

Provenance of the classification (e.g. "aws_instance_state").
termination_confidence str | None

How sure the producer was ("high" / "low").
cost_estimate float | None

Estimated USD cost of the job (list/market price, not billed).
Source code in snakesee/events.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@dataclass(frozen=True, slots=True)
class SnakeseeEvent:
    """A single event from the logger plugin.

    This is a frozen dataclass to ensure events are immutable once parsed.

    Attributes:
        event_type: Type of the event.
        timestamp: Unix timestamp when the event occurred.
        job_id: Snakemake job ID (for job events).
        rule_name: Name of the rule (for job events).
        wildcards: Wildcard values for the job.
        threads: Number of threads allocated to the job.
        resources: Resource requirements for the job.
        input_files: Tuple of input file paths.
        output_files: Tuple of output file paths.
        duration: Job duration in seconds (for finished/error events).
        error_message: Error message (for error events).
        completed_jobs: Number of completed jobs (for progress events).
        total_jobs: Total number of jobs (for progress events).
        workflow_id: Unique workflow identifier.
        executor: Remote executor identifier (e.g. "aws-batch"), if applicable.
        external_jobid: External executor job id/ARN, for remote jobs.
        remote_status: Raw backend status string (e.g. "RUNNING"), for remote jobs.
        queued_at: Epoch seconds the job entered the remote queue, if known.
        started_at: Epoch seconds the job began executing on a remote node, if known.
        stopped_at: Epoch seconds the job stopped executing remotely, if known.
        attempt: 1-based attempt number for retried/preempted remote jobs.
        exit_code: Container/process exit code for a finished remote job.
        status_reason: Backend-provided reason string (e.g. failure cause).
        queue: The remote queue the job was routed to, if known.
        log_stream: Backend log stream identifier (e.g. CloudWatch stream).
        region: Cloud region, used to build console deep links.
        termination_category: Why the job died (e.g. "spot", "oom"), if classified.
        termination_source: Provenance of the classification (e.g. "aws_instance_state").
        termination_confidence: How sure the producer was ("high" / "low").
        cost_estimate: Estimated USD cost of the job (list/market price, not billed).
    """

    event_type: EventType
    timestamp: float
    job_id: int | None = None
    rule_name: str | None = None
    wildcards: tuple[tuple[str, str], ...] | None = None
    threads: int | None = None
    resources: tuple[tuple[str, Any], ...] | None = None
    input_files: tuple[str, ...] | None = None
    output_files: tuple[str, ...] | None = None
    duration: float | None = None
    error_message: str | None = None
    completed_jobs: int | None = None
    total_jobs: int | None = None
    workflow_id: str | None = None
    # Remote-executor enrichment (all optional; absent for local jobs).
    executor: str | None = None
    external_jobid: str | None = None
    remote_status: str | None = None
    queued_at: float | None = None
    started_at: float | None = None
    stopped_at: float | None = None
    attempt: int | None = None
    exit_code: int | None = None
    status_reason: str | None = None
    queue: str | None = None
    log_stream: str | None = None
    region: str | None = None
    termination_category: str | None = None
    termination_source: str | None = None
    termination_confidence: str | None = None
    cost_estimate: float | None = None

    @classmethod
    def from_json(cls, json_str: str | bytes) -> "SnakeseeEvent":
        """Parse from JSON line.

        Args:
            json_str: JSON string or bytes to parse.

        Returns:
            Parsed SnakeseeEvent instance.

        Raises:
            ValueError: If the JSON is invalid or has an unknown event type.
            orjson.JSONDecodeError: If the JSON cannot be parsed.
        """
        data = orjson.loads(json_str)
        data["event_type"] = EventType(data["event_type"])

        # Convert dicts to tuples for frozen dataclass compatibility
        if "wildcards" in data and data["wildcards"] is not None:
            data["wildcards"] = tuple(sorted(data["wildcards"].items()))
        if "resources" in data and data["resources"] is not None:
            data["resources"] = tuple(sorted(data["resources"].items()))
        if "input_files" in data and data["input_files"] is not None:
            data["input_files"] = tuple(data["input_files"])
        if "output_files" in data and data["output_files"] is not None:
            data["output_files"] = tuple(data["output_files"])

        # Drop unknown keys for forward compatibility: a newer plugin may emit
        # fields this build of snakesee doesn't model yet, and cls(**data) would
        # otherwise raise on the unexpected keyword. Skipping them lets an older
        # reader keep working against a newer event stream.
        known = {f.name for f in fields(cls)}
        data = {k: v for k, v in data.items() if k in known}

        return cls(**data)

    @property
    def wildcards_dict(self) -> dict[str, str] | None:
        """Get wildcards as a dictionary.

        Returns:
            Wildcards as a dict, or None if not set.
        """
        if self.wildcards is None:
            return None
        return dict(self.wildcards)

Attributes

wildcards_dict property 
wildcards_dict: dict[str, str] | None

Get wildcards as a dictionary.

Returns:

Type Description
dict[str, str] | None

Wildcards as a dict, or None if not set.

Methods:

from_json classmethod 
from_json(json_str: str | bytes) -> SnakeseeEvent

Parse from JSON line.

Parameters:

Name Type Description Default
json_str str | bytes

JSON string or bytes to parse.

 required

Returns:

Type Description
SnakeseeEvent

Parsed SnakeseeEvent instance.

Raises:

Type Description
ValueError

If the JSON is invalid or has an unknown event type.
JSONDecodeError

If the JSON cannot be parsed.
Source code in snakesee/events.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def from_json(cls, json_str: str | bytes) -> "SnakeseeEvent":
    """Parse from JSON line.

    Args:
        json_str: JSON string or bytes to parse.

    Returns:
        Parsed SnakeseeEvent instance.

    Raises:
        ValueError: If the JSON is invalid or has an unknown event type.
        orjson.JSONDecodeError: If the JSON cannot be parsed.
    """
    data = orjson.loads(json_str)
    data["event_type"] = EventType(data["event_type"])

    # Convert dicts to tuples for frozen dataclass compatibility
    if "wildcards" in data and data["wildcards"] is not None:
        data["wildcards"] = tuple(sorted(data["wildcards"].items()))
    if "resources" in data and data["resources"] is not None:
        data["resources"] = tuple(sorted(data["resources"].items()))
    if "input_files" in data and data["input_files"] is not None:
        data["input_files"] = tuple(data["input_files"])
    if "output_files" in data and data["output_files"] is not None:
        data["output_files"] = tuple(data["output_files"])

    # Drop unknown keys for forward compatibility: a newer plugin may emit
    # fields this build of snakesee doesn't model yet, and cls(**data) would
    # otherwise raise on the unexpected keyword. Skipping them lets an older
    # reader keep working against a newer event stream.
    known = {f.name for f in fields(cls)}
    data = {k: v for k, v in data.items() if k in known}

    return cls(**data)

Functions:

get_event_file_path 

get_event_file_path(workflow_dir: Path) -> Path

Get the path to the event file for a workflow.

Parameters:

Name Type Description Default
workflow_dir Path

Path to the workflow directory.

 required

Returns:

Type Description
Path

Path to the event file.
Source code in snakesee/events.py 
242
243
244
245
246
247
248
249
250
251
def get_event_file_path(workflow_dir: Path) -> Path:
    """Get the path to the event file for a workflow.

    Args:
        workflow_dir: Path to the workflow directory.

    Returns:
        Path to the event file.
    """
    return workflow_dir / EVENT_FILE_NAME