Skip to content

job_tracker

Job lifecycle tracking from start to completion.

Classes

JobLifecycleTracker

Tracks jobs from start to completion.

Maintains state for started jobs, finished job IDs, and completed jobs with full timing information.

Source code in snakesee/parser/job_tracker.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 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
156
157
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
class JobLifecycleTracker:
    """Tracks jobs from start to completion.

    Maintains state for started jobs, finished job IDs, and completed
    jobs with full timing information.
    """

    __slots__ = (
        "_completed_jobs",
        "_finished_jobids",
        "_job_logs",
        "_started_jobs",
    )

    def __init__(self) -> None:
        """Initialize the job tracker."""
        self._started_jobs: dict[str, StartedJobData] = {}
        self._finished_jobids: set[str] = set()
        self._completed_jobs: list[JobInfo] = []
        self._job_logs: dict[str, str] = {}

    def start_job(
        self,
        jobid: str,
        rule: str,
        start_time: float | None = None,
        wildcards: dict[str, str] | None = None,
        threads: int | None = None,
    ) -> None:
        """Record a job starting.

        Args:
            jobid: Unique job identifier.
            rule: Name of the rule.
            start_time: Unix timestamp when job started.
            wildcards: Wildcard values for this job.
            threads: Number of threads allocated.
        """
        if jobid not in self._started_jobs:
            self._started_jobs[jobid] = StartedJobData(
                rule=rule,
                start_time=start_time,
                wildcards=wildcards,
                threads=threads,
            )

    def update_job(
        self,
        jobid: str,
        wildcards: dict[str, str] | None = None,
        threads: int | None = None,
    ) -> None:
        """Update an existing job's metadata.

        Args:
            jobid: Job identifier.
            wildcards: Wildcard values to update.
            threads: Thread count to update.
        """
        if jobid in self._started_jobs:
            if wildcards is not None:
                self._started_jobs[jobid]["wildcards"] = wildcards
            if threads is not None:
                self._started_jobs[jobid]["threads"] = threads

    def set_job_log(self, jobid: str, log_path: str) -> None:
        """Set the log file path for a job.

        Args:
            jobid: Job identifier.
            log_path: Path to the log file.
        """
        self._job_logs[jobid] = log_path

    def get_job_log(self, jobid: str) -> str | None:
        """Get the log file path for a job.

        Args:
            jobid: Job identifier.

        Returns:
            Log file path or None if not set.
        """
        return self._job_logs.get(jobid)

    def finish_job(self, jobid: str, end_time: float | None = None) -> JobInfo | None:
        """Record a job finishing.

        Args:
            jobid: Job identifier.
            end_time: Unix timestamp when job finished.

        Returns:
            JobInfo for the completed job, or None if job was not tracked.
        """
        self._finished_jobids.add(jobid)

        if jobid in self._started_jobs:
            job_data = self._started_jobs[jobid]
            log_path = self._job_logs.get(jobid)
            job_info = JobInfo(
                rule=job_data["rule"],
                job_id=jobid,
                start_time=job_data["start_time"],
                end_time=end_time,
                wildcards=job_data["wildcards"],
                threads=job_data["threads"],
                log_file=Path(log_path) if log_path else None,
            )
            self._completed_jobs.append(job_info)
            # Clean up started job data to prevent memory growth
            del self._started_jobs[jobid]
            # Also clean up the job log entry
            self._job_logs.pop(jobid, None)
            return job_info
        return None

    def get_running_jobs(self) -> list[JobInfo]:
        """Get list of jobs that started but haven't finished.

        Returns:
            List of JobInfo for running jobs.
        """
        running: list[JobInfo] = []
        for jobid, job_data in self._started_jobs.items():
            if jobid not in self._finished_jobids:
                log_path = self._job_logs.get(jobid)
                running.append(
                    JobInfo(
                        rule=job_data["rule"],
                        job_id=jobid,
                        start_time=job_data["start_time"],
                        wildcards=job_data["wildcards"],
                        threads=job_data["threads"],
                        log_file=Path(log_path) if log_path else None,
                    )
                )
        return running

    def get_completed_jobs(self) -> list[JobInfo]:
        """Get list of completed jobs sorted by end time (newest first).

        Returns:
            List of JobInfo for completed jobs.
        """
        return sorted(
            self._completed_jobs,
            key=lambda j: j.end_time or 0,
            reverse=True,
        )

    def is_job_started(self, jobid: str) -> bool:
        """Check if a job has been started.

        Args:
            jobid: Job identifier.

        Returns:
            True if the job has been started.
        """
        return jobid in self._started_jobs

    def reset(self) -> None:
        """Clear all job tracking state."""
        self._started_jobs.clear()
        self._finished_jobids.clear()
        self._completed_jobs.clear()
        self._job_logs.clear()

Functions

__init__ 
__init__() -> None

Initialize the job tracker.

Source code in snakesee/parser/job_tracker.py 
32
33
34
35
36
37
def __init__(self) -> None:
    """Initialize the job tracker."""
    self._started_jobs: dict[str, StartedJobData] = {}
    self._finished_jobids: set[str] = set()
    self._completed_jobs: list[JobInfo] = []
    self._job_logs: dict[str, str] = {}
finish_job 
finish_job(jobid: str, end_time: float | None = None) -> JobInfo | None

Record a job finishing.

Parameters:

Name Type Description Default
jobid str

Job identifier.

 required
end_time float | None

Unix timestamp when job finished.

 None

Returns:

Type Description
JobInfo | None

JobInfo for the completed job, or None if job was not tracked.
Source code in snakesee/parser/job_tracker.py 
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
def finish_job(self, jobid: str, end_time: float | None = None) -> JobInfo | None:
    """Record a job finishing.

    Args:
        jobid: Job identifier.
        end_time: Unix timestamp when job finished.

    Returns:
        JobInfo for the completed job, or None if job was not tracked.
    """
    self._finished_jobids.add(jobid)

    if jobid in self._started_jobs:
        job_data = self._started_jobs[jobid]
        log_path = self._job_logs.get(jobid)
        job_info = JobInfo(
            rule=job_data["rule"],
            job_id=jobid,
            start_time=job_data["start_time"],
            end_time=end_time,
            wildcards=job_data["wildcards"],
            threads=job_data["threads"],
            log_file=Path(log_path) if log_path else None,
        )
        self._completed_jobs.append(job_info)
        # Clean up started job data to prevent memory growth
        del self._started_jobs[jobid]
        # Also clean up the job log entry
        self._job_logs.pop(jobid, None)
        return job_info
    return None
get_completed_jobs 
get_completed_jobs() -> list[JobInfo]

Get list of completed jobs sorted by end time (newest first).

Returns:

Type Description
list[JobInfo]

List of JobInfo for completed jobs.
Source code in snakesee/parser/job_tracker.py 
157
158
159
160
161
162
163
164
165
166
167
def get_completed_jobs(self) -> list[JobInfo]:
    """Get list of completed jobs sorted by end time (newest first).

    Returns:
        List of JobInfo for completed jobs.
    """
    return sorted(
        self._completed_jobs,
        key=lambda j: j.end_time or 0,
        reverse=True,
    )
get_job_log 
get_job_log(jobid: str) -> str | None

Get the log file path for a job.

Parameters:

Name Type Description Default
jobid str

Job identifier.

 required

Returns:

Type Description
str | None

Log file path or None if not set.
Source code in snakesee/parser/job_tracker.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_job_log(self, jobid: str) -> str | None:
    """Get the log file path for a job.

    Args:
        jobid: Job identifier.

    Returns:
        Log file path or None if not set.
    """
    return self._job_logs.get(jobid)
get_running_jobs 
get_running_jobs() -> list[JobInfo]

Get list of jobs that started but haven't finished.

Returns:

Type Description
list[JobInfo]

List of JobInfo for running jobs.
Source code in snakesee/parser/job_tracker.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def get_running_jobs(self) -> list[JobInfo]:
    """Get list of jobs that started but haven't finished.

    Returns:
        List of JobInfo for running jobs.
    """
    running: list[JobInfo] = []
    for jobid, job_data in self._started_jobs.items():
        if jobid not in self._finished_jobids:
            log_path = self._job_logs.get(jobid)
            running.append(
                JobInfo(
                    rule=job_data["rule"],
                    job_id=jobid,
                    start_time=job_data["start_time"],
                    wildcards=job_data["wildcards"],
                    threads=job_data["threads"],
                    log_file=Path(log_path) if log_path else None,
                )
            )
    return running
is_job_started 
is_job_started(jobid: str) -> bool

Check if a job has been started.

Parameters:

Name Type Description Default
jobid str

Job identifier.

 required

Returns:

Type Description
bool

True if the job has been started.
Source code in snakesee/parser/job_tracker.py 
169
170
171
172
173
174
175
176
177
178
def is_job_started(self, jobid: str) -> bool:
    """Check if a job has been started.

    Args:
        jobid: Job identifier.

    Returns:
        True if the job has been started.
    """
    return jobid in self._started_jobs
reset 
reset() -> None

Clear all job tracking state.

Source code in snakesee/parser/job_tracker.py 
180
181
182
183
184
185
def reset(self) -> None:
    """Clear all job tracking state."""
    self._started_jobs.clear()
    self._finished_jobids.clear()
    self._completed_jobs.clear()
    self._job_logs.clear()
set_job_log 
set_job_log(jobid: str, log_path: str) -> None

Set the log file path for a job.

Parameters:

Name Type Description Default
jobid str

Job identifier.

 required
log_path str

Path to the log file.

 required
Source code in snakesee/parser/job_tracker.py 
83
84
85
86
87
88
89
90
def set_job_log(self, jobid: str, log_path: str) -> None:
    """Set the log file path for a job.

    Args:
        jobid: Job identifier.
        log_path: Path to the log file.
    """
    self._job_logs[jobid] = log_path
start_job 
start_job(jobid: str, rule: str, start_time: float | None = None, wildcards: dict[str, str] | None = None, threads: int | None = None) -> None

Record a job starting.

Parameters:

Name Type Description Default
jobid str

Unique job identifier.

 required
rule str

Name of the rule.

 required
start_time float | None

Unix timestamp when job started.

 None
wildcards dict[str, str] | None

Wildcard values for this job.

 None
threads int | None

Number of threads allocated.

 None
Source code in snakesee/parser/job_tracker.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
def start_job(
    self,
    jobid: str,
    rule: str,
    start_time: float | None = None,
    wildcards: dict[str, str] | None = None,
    threads: int | None = None,
) -> None:
    """Record a job starting.

    Args:
        jobid: Unique job identifier.
        rule: Name of the rule.
        start_time: Unix timestamp when job started.
        wildcards: Wildcard values for this job.
        threads: Number of threads allocated.
    """
    if jobid not in self._started_jobs:
        self._started_jobs[jobid] = StartedJobData(
            rule=rule,
            start_time=start_time,
            wildcards=wildcards,
            threads=threads,
        )
update_job 
update_job(jobid: str, wildcards: dict[str, str] | None = None, threads: int | None = None) -> None

Update an existing job's metadata.

Parameters:

Name Type Description Default
jobid str

Job identifier.

 required
wildcards dict[str, str] | None

Wildcard values to update.

 None
threads int | None

Thread count to update.

 None
Source code in snakesee/parser/job_tracker.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def update_job(
    self,
    jobid: str,
    wildcards: dict[str, str] | None = None,
    threads: int | None = None,
) -> None:
    """Update an existing job's metadata.

    Args:
        jobid: Job identifier.
        wildcards: Wildcard values to update.
        threads: Thread count to update.
    """
    if jobid in self._started_jobs:
        if wildcards is not None:
            self._started_jobs[jobid]["wildcards"] = wildcards
        if threads is not None:
            self._started_jobs[jobid]["threads"] = threads

StartedJobData

Bases: TypedDict

Data for a job that has started but not finished.

Source code in snakesee/parser/job_tracker.py 
 9
10
11
12
13
14
15
class StartedJobData(TypedDict):
    """Data for a job that has started but not finished."""

    rule: str
    start_time: float | None
    wildcards: dict[str, str] | None
    threads: int | None