Skip to content

Scheduler

Workflow-based job pipeline. Narrative guide: Scheduler.

Task

pypepper.scheduler.task

Scheduler task data structures.

DEFAULT_RETRY_UNTIL_MAX module-attribute 

DEFAULT_RETRY_UNTIL_MAX = 1000

ITask

Bases: IBase

Source code in pypepper/scheduler/task.py 
15
16
17
18
19
20
21
22
class ITask(IBase, metaclass=ABCMeta):
    retry_count: int = 0
    retry_delay: int = 0
    retry_until_completed: bool = False
    # Per-round cap for until-retries when retry_count==0 (see Workflow._run_task).
    retry_until_max: int = DEFAULT_RETRY_UNTIL_MAX
    optional: bool = False
    executor: IExecutor

retry_count class-attribute instance-attribute 

retry_count: int = 0

retry_delay class-attribute instance-attribute 

retry_delay: int = 0

retry_until_completed class-attribute instance-attribute 

retry_until_completed: bool = False

retry_until_max class-attribute instance-attribute 

retry_until_max: int = DEFAULT_RETRY_UNTIL_MAX

optional class-attribute instance-attribute 

optional: bool = False

executor instance-attribute 

executor: IExecutor

Task

Bases: ITask

Source code in pypepper/scheduler/task.py 
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
class Task(ITask):
    def __init__(
        self,
        channel_id: str,
        dag_id: str,
        fingerprint: str,
        name: str,
        category: str,
        description: str,
        tags: list[Tag],
        executor: IExecutor,
        round_timeout: int = 0,
        round_times: int = 1,
        version: int = 1,
        retry_count: int = 0,
        retry_delay: int = 0,
        retry_until_completed: bool = False,
        retry_until_max: int = DEFAULT_RETRY_UNTIL_MAX,
        optional: bool = False,
        executor_id: str | None = None,
        round_timeout_join: int = 0,
    ) -> None:
        if retry_until_max < 1:
            raise ValueError(f"retry_until_max must be >= 1, got {retry_until_max}")
        if round_times < 1:
            raise ValueError(f"round_times must be >= 1, got {round_times}")
        if round_timeout < 0:
            raise ValueError(f"round_timeout must be >= 0, got {round_timeout}")
        if round_timeout_join < 0:
            raise ValueError(f"round_timeout_join must be >= 0, got {round_timeout_join}")
        if retry_count < 0:
            raise ValueError(f"retry_count must be >= 0, got {retry_count}")
        if retry_delay < 0:
            raise ValueError(f"retry_delay must be >= 0, got {retry_delay}")

        self.channel_id = channel_id
        self.dag_id = dag_id
        self.fingerprint = fingerprint
        self.name = name
        self.category = category
        self.description = description
        self.tags = tags
        self.executor = executor
        self.executor_id = executor_id
        # Seconds per execute attempt; 0 = none. Started work is joined on timeout.
        self.round_timeout = round_timeout
        # Extra seconds to join a started execute after round_timeout (0 = wait forever).
        self.round_timeout_join = round_timeout_join
        # Outer rounds; each round has its own inner retry budget.
        self.round_times = round_times
        self.version = version
        self.retry_count = retry_count
        self.retry_delay = retry_delay
        # When True and retry_count==0, retry until success up to retry_until_max per round.
        self.retry_until_completed = retry_until_completed
        self.retry_until_max = retry_until_max
        self.optional = optional
        self.id = uuid.new_uuid()
        self.context = Context(context_id=uuid.new_uuid())

channel_id instance-attribute 

channel_id = channel_id

dag_id instance-attribute 

dag_id = dag_id

fingerprint instance-attribute 

fingerprint = fingerprint

name instance-attribute 

name = name

category instance-attribute 

category = category

description instance-attribute 

description = description

tags instance-attribute 

tags = tags

executor instance-attribute 

executor = executor

executor_id instance-attribute 

executor_id = executor_id

round_timeout instance-attribute 

round_timeout = round_timeout

round_timeout_join instance-attribute 

round_timeout_join = round_timeout_join

round_times instance-attribute 

round_times = round_times

version instance-attribute 

version = version

retry_count instance-attribute 

retry_count = retry_count

retry_delay instance-attribute 

retry_delay = retry_delay

retry_until_completed instance-attribute 

retry_until_completed = retry_until_completed

retry_until_max instance-attribute 

retry_until_max = retry_until_max

optional instance-attribute 

optional = optional

id instance-attribute 

id = uuid.new_uuid()

context instance-attribute 

context = Context(context_id=uuid.new_uuid())

Workflow

pypepper.scheduler.workflow

Sequential workflow runner over tasks.

IWorkflow

Bases: IBase

Source code in pypepper/scheduler/workflow.py 
42
43
class IWorkflow(IBase, metaclass=ABCMeta):
    tasks: list[Task]

tasks instance-attribute 

tasks: list[Task]

Workflow

Bases: IWorkflow

Source code in pypepper/scheduler/workflow.py 
 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
186
187
188
189
190
191
class Workflow(IWorkflow):
    def __init__(self) -> None:
        self.tasks: list[Task] = []

    def add_task(self, task: Task) -> None:
        self.tasks.append(task)

    def add_tasks(self, tasks: list[Task]) -> None:
        self.tasks.extend(tasks)

    def get_tasks(self) -> list[Task]:
        return self.tasks

    def run(self) -> list[object]:
        """
        Sequentially execute tasks.

        Per task:
        - ``round_times`` outer rounds (default 1); each round has its own retry budget.
          Success returns early; later rounds run only after a full failed inner budget.
        - ``round_timeout`` seconds per execute attempt (0 = none). Timeout counts as a
          failed attempt. Queued work is cancelled when possible (``timed out before
          start``). Started work is **joined** before the next retry or ``run()``
          return (``execute still running``). ``round_timeout_join`` bounds that
          wait (``0`` = forever); a still-running Future blocks a later attempt
          for the same task. A hung execute with infinite join blocks this workflow /
          Worker. Concurrent timeout executes are capped (``_SOFT_TIMEOUT_MAX_WORKERS``);
          further work queues and a short timeout may fire before the task starts.
        - Retry modes: until false → ``retry_count + 1``; until + count 0 → per-round
          ``retry_until_max``; until + count > 0 → ``retry_count + 1`` (max ignored).
        - ``optional``: failed optional tasks continue the workflow.

        Non-optional task failure after all rounds/attempts aborts the workflow.
        """
        from pypepper.common.tracing import get_tracer

        with get_tracer("pypepper.scheduler").start_as_current_span("scheduler.workflow.run"):
            if len(self.tasks) == 0:
                return []

            results: list[object] = []
            for task in self.tasks:
                result = self._run_task(task)
                results.append(result)
            return results

    @staticmethod
    def _attempts_per_round(task: Task) -> int:
        retry_count = int(task.retry_count or 0)
        if task.retry_until_completed and retry_count == 0:
            return max(1, int(task.retry_until_max))
        return max(1, retry_count + 1)

    @staticmethod
    def _execute_once(task: Task) -> object | None:
        executor = task.executor
        if executor is None:
            return None

        timeout = int(task.round_timeout or 0)
        if timeout <= 0:
            return cast(object | None, executor.execute(task, task.context))

        with _inflight_lock:
            existing = _inflight.get(task.id)
            if existing is not None and not existing.done():
                raise TimeoutError(
                    f"Task execute still running from a previous attempt: id={task.id}, name={task.name}"
                )
            if existing is not None and existing.done():
                _inflight.pop(task.id, None)

        future: Future[object | None] = _soft_timeout_pool().submit(executor.execute, task, task.context)
        with _inflight_lock:
            _inflight[task.id] = future
        try:
            return cast(object | None, future.result(timeout=timeout))
        except FuturesTimeoutError as e:
            if future.done():
                return cast(object | None, future.result())
            if future.cancel():
                with _inflight_lock:
                    _inflight.pop(task.id, None)
                raise TimeoutError(
                    f"Task execute timed out before start "
                    f"(round_timeout={timeout}s, still queued): id={task.id}, name={task.name}"
                ) from e
            if future.done():
                return cast(object | None, future.result())
            join_timeout = int(task.round_timeout_join or 0)
            try:
                if join_timeout > 0:
                    future.result(timeout=join_timeout)
                else:
                    future.result()
            except FuturesTimeoutError as join_exc:
                log.warn(
                    f"Task execute exceeded round_timeout={timeout}s "
                    f"(execute still running after join_timeout={join_timeout}s): "
                    f"id={task.id}, name={task.name}"
                )
                raise TimeoutError(
                    f"Task execute exceeded round_timeout={timeout}s "
                    f"(execute still running): id={task.id}, name={task.name}"
                ) from join_exc
            except Exception as execute_exc:
                log.warn(
                    f"Task execute exceeded round_timeout={timeout}s "
                    f"(execute still running): id={task.id}, name={task.name}, error={execute_exc}"
                )
                raise TimeoutError(
                    f"Task execute exceeded round_timeout={timeout}s "
                    f"(execute still running): id={task.id}, name={task.name}"
                ) from e
            raise TimeoutError(
                f"Task execute exceeded round_timeout={timeout}s "
                f"(execute still running): id={task.id}, name={task.name}"
            ) from e
        finally:
            if future.done():
                with _inflight_lock:
                    _inflight.pop(task.id, None)

    def _run_task(self, task: Task) -> object | None:
        rounds = max(1, int(task.round_times or 1))
        attempts = self._attempts_per_round(task)
        last_error: Exception | None = None

        for round_idx in range(rounds):
            for attempt in range(attempts):
                try:
                    return self._execute_once(task)
                except Exception as e:
                    last_error = e
                    log.warn(
                        f"Task failed: id={task.id}, name={task.name}, "
                        f"round={round_idx + 1}/{rounds}, attempt={attempt + 1}/{attempts}, error={e}"
                    )
                    if attempt + 1 < attempts and task.retry_delay:
                        time.sleep(task.retry_delay)

        if task.optional:
            log.warn(f"Optional task failed, continuing: id={task.id}, error={last_error}")
            return None

        raise last_error if last_error else RuntimeError(f"Task failed: {task.id}")

tasks instance-attribute 

tasks: list[Task] = []

add_task 

add_task(task: Task) -> None
Source code in pypepper/scheduler/workflow.py 
50
51
def add_task(self, task: Task) -> None:
    self.tasks.append(task)

add_tasks 

add_tasks(tasks: list[Task]) -> None
Source code in pypepper/scheduler/workflow.py 
53
54
def add_tasks(self, tasks: list[Task]) -> None:
    self.tasks.extend(tasks)

get_tasks 

get_tasks() -> list[Task]
Source code in pypepper/scheduler/workflow.py 
56
57
def get_tasks(self) -> list[Task]:
    return self.tasks

run 

run() -> list[object]

Sequentially execute tasks.

Per task: - round_times outer rounds (default 1); each round has its own retry budget. Success returns early; later rounds run only after a full failed inner budget. - round_timeout seconds per execute attempt (0 = none). Timeout counts as a failed attempt. Queued work is cancelled when possible (timed out before start). Started work is joined before the next retry or run() return (execute still running). round_timeout_join bounds that wait (0 = forever); a still-running Future blocks a later attempt for the same task. A hung execute with infinite join blocks this workflow / Worker. Concurrent timeout executes are capped (_SOFT_TIMEOUT_MAX_WORKERS); further work queues and a short timeout may fire before the task starts. - Retry modes: until false → retry_count + 1; until + count 0 → per-round retry_until_max; until + count > 0 → retry_count + 1 (max ignored). - optional: failed optional tasks continue the workflow.

Non-optional task failure after all rounds/attempts aborts the workflow.

Source code in pypepper/scheduler/workflow.py 
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
def run(self) -> list[object]:
    """
    Sequentially execute tasks.

    Per task:
    - ``round_times`` outer rounds (default 1); each round has its own retry budget.
      Success returns early; later rounds run only after a full failed inner budget.
    - ``round_timeout`` seconds per execute attempt (0 = none). Timeout counts as a
      failed attempt. Queued work is cancelled when possible (``timed out before
      start``). Started work is **joined** before the next retry or ``run()``
      return (``execute still running``). ``round_timeout_join`` bounds that
      wait (``0`` = forever); a still-running Future blocks a later attempt
      for the same task. A hung execute with infinite join blocks this workflow /
      Worker. Concurrent timeout executes are capped (``_SOFT_TIMEOUT_MAX_WORKERS``);
      further work queues and a short timeout may fire before the task starts.
    - Retry modes: until false → ``retry_count + 1``; until + count 0 → per-round
      ``retry_until_max``; until + count > 0 → ``retry_count + 1`` (max ignored).
    - ``optional``: failed optional tasks continue the workflow.

    Non-optional task failure after all rounds/attempts aborts the workflow.
    """
    from pypepper.common.tracing import get_tracer

    with get_tracer("pypepper.scheduler").start_as_current_span("scheduler.workflow.run"):
        if len(self.tasks) == 0:
            return []

        results: list[object] = []
        for task in self.tasks:
            result = self._run_task(task)
            results.append(result)
        return results

Job

pypepper.scheduler.job

Job model, processor, and dispatcher.

CANCEL_EVENT_KEY module-attribute 

CANCEL_EVENT_KEY = 'pypepper.scheduler.cancel_event'

dispatcher module-attribute 

dispatcher = Dispatcher()

ChannelEnqueueError

Bases: RuntimeError

Enqueue rejected before the job lands on the channel (safe to roll back).

Source code in pypepper/scheduler/job.py 
28
29
class ChannelEnqueueError(RuntimeError):
    """Enqueue rejected before the job lands on the channel (safe to roll back)."""

ChannelFullError

Bases: ChannelEnqueueError

Bounded channel capacity rejection (pre-execution; safe to roll back).

Source code in pypepper/scheduler/job.py 
32
33
class ChannelFullError(ChannelEnqueueError):
    """Bounded channel capacity rejection (pre-execution; safe to roll back)."""

ChannelStoppedError

Bases: ChannelEnqueueError

Channel was stopped; enqueue rejected (pre-execution; safe to roll back).

Sibling of :class:ChannelFullError (not a subclass): catch this before treating :class:ChannelFullError as retryable backpressure. Prefer catching :class:ChannelEnqueueError for any pre-landing rejection.

Source code in pypepper/scheduler/job.py 
36
37
38
39
40
41
42
43
class ChannelStoppedError(ChannelEnqueueError):
    """
    Channel was stopped; enqueue rejected (pre-execution; safe to roll back).

    Sibling of :class:`ChannelFullError` (not a subclass): catch this before treating
    :class:`ChannelFullError` as retryable backpressure. Prefer catching
    :class:`ChannelEnqueueError` for any pre-landing rejection.
    """

JobRedeliveryError

Bases: RuntimeError

Dequeued job could not be returned to the channel after a RUN-start restore (channel full or stopped). Worker.run_forever re-raises. When reason is stopped, leftover queued jobs are drained first; full stops immediately. job is the unrestored instance (not on the channel).

Source code in pypepper/scheduler/job.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class JobRedeliveryError(RuntimeError):
    """
    Dequeued job could not be returned to the channel after a RUN-start restore
    (channel full or stopped). ``Worker.run_forever`` re-raises. When ``reason`` is
    ``stopped``, leftover queued jobs are drained first; ``full`` stops immediately.
    ``job`` is the unrestored instance (not on the channel).
    """

    def __init__(self, message: str, *, reason: str, job: Job | None = None) -> None:
        super().__init__(message)
        if reason not in ("full", "stopped"):
            raise ValueError(f"JobRedeliveryError.reason must be 'full' or 'stopped', got {reason!r}")
        self.reason = reason
        self.job = job

reason instance-attribute 

reason = reason

job instance-attribute 

job = job

JobRequeuedError

Bases: RuntimeError

Job was restored and put back on the channel after RUN-start persist failure.

Worker.run_forever re-raises (does not continue) so supervisors see a non-success exit and avoid persist-failure busy-spin; the job remains queued for a later consumer.

Source code in pypepper/scheduler/job.py 
62
63
64
65
66
67
68
class JobRequeuedError(RuntimeError):
    """
    Job was restored and put back on the channel after RUN-start persist failure.

    ``Worker.run_forever`` re-raises (does not continue) so supervisors see a non-success
    exit and avoid persist-failure busy-spin; the job remains queued for a later consumer.
    """

Processor

Source code in pypepper/scheduler/job.py 
 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
class Processor:
    def run(self, job: Job, chan: Channel, *, on_enqueued: Callable[[], None] | None = None) -> None:
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            asyncio.run(self.async_run(job, chan, on_enqueued=on_enqueued))
            return
        raise RuntimeError(
            "Processor.run / Job.scheduled() must be called from a sync context "
            "(no running event loop); from async code apply INIT→SCHEDULE, "
            "await Channel.send(job) (returns ok/full/stopped), then job.save(), "
            "and consume with Worker"
        )

    @staticmethod
    async def async_run(
        job: Job,
        chan: Channel,
        *,
        on_enqueued: Callable[[], None] | None = None,
    ) -> None:
        result = await chan.send(job)
        if result == SEND_STOPPED:
            raise ChannelStoppedError(f"channel stopped: channel_id={job.channel_id}, job_id={job.id}")
        if result != SEND_OK:
            raise ChannelFullError(f"channel full: channel_id={job.channel_id}, job_id={job.id}")
        # Job is on the channel: callers must not roll back schedule/store after this.
        if on_enqueued is not None:
            on_enqueued()
        log.debug(f"Job enqueued: id={job.id}, channel_length={chan.length()}")

run 

run(
    job: Job,
    chan: Channel,
    *,
    on_enqueued: Callable[[], None] | None = None,
) -> None
Source code in pypepper/scheduler/job.py 
80
81
82
83
84
85
86
87
88
89
90
91
def run(self, job: Job, chan: Channel, *, on_enqueued: Callable[[], None] | None = None) -> None:
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        asyncio.run(self.async_run(job, chan, on_enqueued=on_enqueued))
        return
    raise RuntimeError(
        "Processor.run / Job.scheduled() must be called from a sync context "
        "(no running event loop); from async code apply INIT→SCHEDULE, "
        "await Channel.send(job) (returns ok/full/stopped), then job.save(), "
        "and consume with Worker"
    )

async_run async staticmethod 

async_run(
    job: Job,
    chan: Channel,
    *,
    on_enqueued: Callable[[], None] | None = None,
) -> None
Source code in pypepper/scheduler/job.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@staticmethod
async def async_run(
    job: Job,
    chan: Channel,
    *,
    on_enqueued: Callable[[], None] | None = None,
) -> None:
    result = await chan.send(job)
    if result == SEND_STOPPED:
        raise ChannelStoppedError(f"channel stopped: channel_id={job.channel_id}, job_id={job.id}")
    if result != SEND_OK:
        raise ChannelFullError(f"channel full: channel_id={job.channel_id}, job_id={job.id}")
    # Job is on the channel: callers must not roll back schedule/store after this.
    if on_enqueued is not None:
        on_enqueued()
    log.debug(f"Job enqueued: id={job.id}, channel_length={chan.length()}")

Dispatcher

Source code in pypepper/scheduler/job.py 
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
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
class Dispatcher:
    _instance: Dispatcher | None = None
    _init_lock = Lock()
    _lock: Lock
    _processors: MutableMapping[str, Processor]

    def __new__(cls) -> Dispatcher:
        with cls._init_lock:
            if cls._instance is None:
                inst = super().__new__(cls)
                inst._lock = Lock()
                inst._processors = {}
                cls._instance = inst
            return cls._instance

    def __init__(self) -> None:
        pass

    def _put_processor(self, key: str, processor: Processor) -> None:
        if not key:
            raise ValueError("invalid key")
        if processor is None:
            raise ValueError("invalid processor")

        with self._lock:
            self._processors[key] = processor

    def _get_processor(self, key: str) -> Processor | None:
        if not key:
            raise ValueError("invalid key")

        with self._lock:
            if len(self._processors) == 0:
                return None

            return self._processors.get(key)

    def _new_processor(self, key: str) -> Processor:
        processor = self._get_processor(key)
        if processor is None:
            processor = Processor()
            self._put_processor(key, processor)

        return processor

    def _available_processor(self, key: str) -> Processor:
        return self._new_processor(key)

    def dispatch(self, job: Job) -> None:
        # Pre-channel apply/enqueue failure must roll back so retry can re-enter.
        # Persist Scheduled only after a successful send so enqueue failure cannot
        # leave a store row (no cleanup delete / ghost).
        prev_state = job._fsm.current()
        prev_status = job.status
        try:
            job.apply_event(events.INIT)
            job.apply_event(events.SCHEDULE)
        except Exception as exc:
            job.restore_lifecycle(prev_state, prev_status)
            log.error(f"Job schedule failed: id={job.id}, error={exc}")
            raise

        job.log()

        # Setup + enqueue: roll back only if the job never landed on the channel.
        # After successful send, persist Scheduled; exceptions are committed-enqueue
        # plus secondary failure (including a failed save after send).
        enqueued = False

        def _mark_enqueued() -> None:
            nonlocal enqueued
            enqueued = True

        try:
            chan = manager.available(job.channel_id)
            processor = self._available_processor(job.channel_id)
            processor.run(job, chan, on_enqueued=_mark_enqueued)
            last_save_exc: Exception | None = None
            applied = False
            for _attempt in range(_DISPATCH_SAVE_ATTEMPTS):
                try:
                    if job.save():
                        applied = True
                        last_save_exc = None
                        break
                except Exception as save_exc:
                    last_save_exc = save_exc
            if not applied:
                if last_save_exc is not None:
                    raise last_save_exc
                raise RuntimeError(
                    f"Job Scheduled persist skipped after enqueue: id={job.id}, channel_id={job.channel_id}"
                )
        except Exception as enqueue_exc:
            if enqueued:
                log.error(
                    f"Job post-enqueue error (committed; job may still run): id={job.id}, "
                    f"channel_id={job.channel_id}, error={enqueue_exc}"
                )
                raise
            job.restore_lifecycle(prev_state, prev_status)
            log.error(f"Job enqueue failed: id={job.id}, channel_id={job.channel_id}, error={enqueue_exc}")
            raise

dispatch 

dispatch(job: Job) -> None
Source code in pypepper/scheduler/job.py 
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
def dispatch(self, job: Job) -> None:
    # Pre-channel apply/enqueue failure must roll back so retry can re-enter.
    # Persist Scheduled only after a successful send so enqueue failure cannot
    # leave a store row (no cleanup delete / ghost).
    prev_state = job._fsm.current()
    prev_status = job.status
    try:
        job.apply_event(events.INIT)
        job.apply_event(events.SCHEDULE)
    except Exception as exc:
        job.restore_lifecycle(prev_state, prev_status)
        log.error(f"Job schedule failed: id={job.id}, error={exc}")
        raise

    job.log()

    # Setup + enqueue: roll back only if the job never landed on the channel.
    # After successful send, persist Scheduled; exceptions are committed-enqueue
    # plus secondary failure (including a failed save after send).
    enqueued = False

    def _mark_enqueued() -> None:
        nonlocal enqueued
        enqueued = True

    try:
        chan = manager.available(job.channel_id)
        processor = self._available_processor(job.channel_id)
        processor.run(job, chan, on_enqueued=_mark_enqueued)
        last_save_exc: Exception | None = None
        applied = False
        for _attempt in range(_DISPATCH_SAVE_ATTEMPTS):
            try:
                if job.save():
                    applied = True
                    last_save_exc = None
                    break
            except Exception as save_exc:
                last_save_exc = save_exc
        if not applied:
            if last_save_exc is not None:
                raise last_save_exc
            raise RuntimeError(
                f"Job Scheduled persist skipped after enqueue: id={job.id}, channel_id={job.channel_id}"
            )
    except Exception as enqueue_exc:
        if enqueued:
            log.error(
                f"Job post-enqueue error (committed; job may still run): id={job.id}, "
                f"channel_id={job.channel_id}, error={enqueue_exc}"
            )
            raise
        job.restore_lifecycle(prev_state, prev_status)
        log.error(f"Job enqueue failed: id={job.id}, channel_id={job.channel_id}, error={enqueue_exc}")
        raise

IJob

Bases: IBase

Source code in pypepper/scheduler/job.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class IJob(IBase, metaclass=ABCMeta):
    workflows: list[Workflow]

    @abstractmethod
    def save(self) -> bool:
        pass

    @abstractmethod
    def log(self) -> None:
        pass

    @abstractmethod
    def scheduled(self) -> None:
        pass

    @abstractmethod
    def cancel(self) -> None:
        pass

    @abstractmethod
    def is_cancelled(self) -> bool:
        pass

workflows instance-attribute 

workflows: list[Workflow]

save abstractmethod 

save() -> bool
Source code in pypepper/scheduler/job.py 
222
223
224
@abstractmethod
def save(self) -> bool:
    pass

log abstractmethod 

log() -> None
Source code in pypepper/scheduler/job.py 
226
227
228
@abstractmethod
def log(self) -> None:
    pass

scheduled abstractmethod 

scheduled() -> None
Source code in pypepper/scheduler/job.py 
230
231
232
@abstractmethod
def scheduled(self) -> None:
    pass

cancel abstractmethod 

cancel() -> None
Source code in pypepper/scheduler/job.py 
234
235
236
@abstractmethod
def cancel(self) -> None:
    pass

is_cancelled abstractmethod 

is_cancelled() -> bool
Source code in pypepper/scheduler/job.py 
238
239
240
@abstractmethod
def is_cancelled(self) -> bool:
    pass

Job

Bases: IJob

Source code in pypepper/scheduler/job.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
class Job(IJob):
    def __init__(self, category: str | None = None, channel_id: str = "default") -> None:
        now = get_utc_datetime()
        self.id = uuid.new_uuid()
        self.category: str | None = category
        self.channel_id = channel_id
        self.context = Context(context_id=uuid.new_uuid())
        self.workflows: list[Workflow] = []
        self._fsm = events.build_scheduler_fsm()
        self.status: str = Status.UNKNOWN.value
        self.created: str = now
        self.updated: str = now
        self.version: int = 1
        self._cancel_event = Event()
        self.context.with_value(CANCEL_EVENT_KEY, self._cancel_event)

    def share_cancel_event(self) -> None:
        """Put this job's cancel Event on every task context (same object ``execute`` receives)."""
        token = self._cancel_event
        self.context.with_value(CANCEL_EVENT_KEY, token)
        for workflow in self.workflows:
            for task in workflow.tasks:
                task.context.with_value(CANCEL_EVENT_KEY, token)

    def _current_status(self) -> str:
        current = self._fsm.current()
        if current is None:
            return Status.UNKNOWN.value
        value = current.value
        if isinstance(value, Status):
            return value.value
        return str(value)

    def is_cancelled(self) -> bool:
        return self._current_status() == Status.CANCELLED.value

    def restore_lifecycle(self, state: IState | None, status: str) -> None:
        """Restore FSM/`status` after schedule/enqueue failure or RUN-start persist failure."""
        self._fsm.restore(state)
        self.status = status

    def apply_event(self, event: IEvent) -> None:
        """Apply an FSM event or raise if the transition is invalid."""
        resp = self._fsm.on(event)
        _raise_if_transition_failed(resp.error)
        self.status = self._current_status()

    def cancel(self) -> None:
        """
        Cancel a Scheduled or InProgress job and persist Cancelled.

        On ``save()`` failure the FSM stays Cancelled; retry ``job.save()`` only.
        """
        self.share_cancel_event()
        self._cancel_event.set()
        self.apply_event(events.CANCEL)
        self.save()

    def to_record(self) -> JobRecord:
        """Authoritative lifecycle snapshot from the FSM (may lead durable ``status``)."""
        return JobRecord(
            id=self.id,
            category=self.category,
            channel_id=self.channel_id,
            status=self._current_status(),
            created=self.created,
            updated=self.updated,
            workflow_count=len(self.workflows),
            version=self.version,
            payload=self._workflows_payload(),
        )

    def _workflows_payload(self) -> dict[str, Any] | None:
        if not self.workflows:
            return None
        workflows_out: list[dict[str, Any]] = []
        for workflow in self.workflows:
            tasks_out: list[dict[str, Any]] = []
            for task in workflow.tasks:
                if not task.executor_id:
                    return None
                tasks_out.append(
                    {
                        "id": task.id,
                        "channel_id": task.channel_id,
                        "dag_id": task.dag_id,
                        "fingerprint": task.fingerprint,
                        "name": task.name,
                        "category": task.category,
                        "description": task.description,
                        "tags": [{"key": tag.key, "value": tag.value} for tag in task.tags],
                        "executor_id": task.executor_id,
                        "round_timeout": task.round_timeout,
                        "round_timeout_join": task.round_timeout_join,
                        "round_times": task.round_times,
                        "version": task.version,
                        "retry_count": task.retry_count,
                        "retry_delay": task.retry_delay,
                        "retry_until_completed": task.retry_until_completed,
                        "retry_until_max": task.retry_until_max,
                        "optional": task.optional,
                    }
                )
            workflows_out.append({"tasks": tasks_out})
        return {"workflows": workflows_out}

    @staticmethod
    def _workflows_from_payload(payload: dict[str, Any]) -> list[Workflow]:
        from pypepper.scheduler.executor import executor_registry
        from pypepper.scheduler.tag import Tag
        from pypepper.scheduler.task import Task
        from pypepper.scheduler.workflow import Workflow

        workflows: list[Workflow] = []
        for wf_spec in payload.get("workflows") or []:
            workflow = Workflow()
            for spec in wf_spec.get("tasks") or []:
                executor_id = spec.get("executor_id")
                if not executor_id:
                    raise ValueError("payload task missing executor_id")
                tags = [Tag(key=str(t.get("key", "")), value=str(t.get("value", ""))) for t in spec.get("tags") or []]
                task = Task(
                    channel_id=str(spec["channel_id"]),
                    dag_id=str(spec["dag_id"]),
                    fingerprint=str(spec["fingerprint"]),
                    name=str(spec["name"]),
                    category=str(spec["category"]),
                    description=str(spec.get("description") or ""),
                    tags=tags,
                    executor=executor_registry.resolve(str(executor_id)),
                    round_timeout=int(spec.get("round_timeout") or 0),
                    round_timeout_join=int(spec.get("round_timeout_join") or 0),
                    round_times=int(spec.get("round_times") or 1),
                    version=int(spec.get("version") or 1),
                    retry_count=int(spec.get("retry_count") or 0),
                    retry_delay=int(spec.get("retry_delay") or 0),
                    retry_until_completed=bool(spec.get("retry_until_completed") or False),
                    retry_until_max=int(spec.get("retry_until_max") or 1000),
                    optional=bool(spec.get("optional") or False),
                    executor_id=str(executor_id),
                )
                if spec.get("id"):
                    task.id = str(spec["id"])
                workflow.add_task(task)
            workflows.append(workflow)
        return workflows

    @classmethod
    def from_record(cls, record: JobRecord) -> Job:
        """
        Rebuild a Job from a store snapshot. Requires registered ``executor_id``s when payload is set.

        Leaves the FSM at Unknown (does not restore ``record.status``) so ``scheduled()`` can run again.
        """
        job = cls(category=record.category, channel_id=record.channel_id)
        job.id = record.id
        job.created = record.created
        job.updated = record.updated
        job.version = record.version
        if record.payload:
            job.workflows = cls._workflows_from_payload(record.payload)
        return job

    def save(self) -> bool:
        from pypepper.common.config import config as app_config
        from pypepper.scheduler.store.memory import InMemoryJobStore

        app_config.ensure_scheduler_job_store_applied(
            using_default_memory_store=isinstance(get_job_store(), InMemoryJobStore)
        )
        status = self._current_status()
        updated = get_utc_datetime()
        record = JobRecord(
            id=self.id,
            category=self.category,
            channel_id=self.channel_id,
            status=status,
            created=self.created,
            updated=updated,
            workflow_count=len(self.workflows),
            version=self.version,
            payload=self._workflows_payload(),
        )
        store = get_job_store()
        applied = store.put(record)
        if not applied:
            log.debug(f"Job save skipped stale snapshot: id={self.id}, attempted={status}, version={self.version}")
            return False
        durable = store.get(self.id)
        if durable is not None:
            self.version = durable.version
        self.updated = updated
        log.debug(f"Job saved: id={self.id}, channel_id={self.channel_id}, status={status}")
        return True

    @staticmethod
    def get_saved(job_id: str) -> JobRecord | None:
        from pypepper.common.config import config as app_config
        from pypepper.scheduler.store.memory import InMemoryJobStore

        app_config.ensure_scheduler_job_store_applied(
            using_default_memory_store=isinstance(get_job_store(), InMemoryJobStore)
        )
        return get_job_store().get(job_id)

    def log(self) -> None:
        log.info(f"Job scheduled: id={self.id}, category={self.category}")

    def scheduled(self) -> None:
        dispatcher.dispatch(self)

id instance-attribute 

id = uuid.new_uuid()

category instance-attribute 

category: str | None = category

channel_id instance-attribute 

channel_id = channel_id

context instance-attribute 

context = Context(context_id=uuid.new_uuid())

workflows instance-attribute 

workflows: list[Workflow] = []

status instance-attribute 

status: str = Status.UNKNOWN.value

created instance-attribute 

created: str = now

updated instance-attribute 

updated: str = now

version instance-attribute 

version: int = 1

share_cancel_event 

share_cancel_event() -> None

Put this job's cancel Event on every task context (same object execute receives).

Source code in pypepper/scheduler/job.py 
259
260
261
262
263
264
265
def share_cancel_event(self) -> None:
    """Put this job's cancel Event on every task context (same object ``execute`` receives)."""
    token = self._cancel_event
    self.context.with_value(CANCEL_EVENT_KEY, token)
    for workflow in self.workflows:
        for task in workflow.tasks:
            task.context.with_value(CANCEL_EVENT_KEY, token)

is_cancelled 

is_cancelled() -> bool
Source code in pypepper/scheduler/job.py 
276
277
def is_cancelled(self) -> bool:
    return self._current_status() == Status.CANCELLED.value

restore_lifecycle 

restore_lifecycle(
    state: IState | None, status: str
) -> None

Restore FSM/status after schedule/enqueue failure or RUN-start persist failure.

Source code in pypepper/scheduler/job.py 
279
280
281
282
def restore_lifecycle(self, state: IState | None, status: str) -> None:
    """Restore FSM/`status` after schedule/enqueue failure or RUN-start persist failure."""
    self._fsm.restore(state)
    self.status = status

apply_event 

apply_event(event: IEvent) -> None

Apply an FSM event or raise if the transition is invalid.

Source code in pypepper/scheduler/job.py 
284
285
286
287
288
def apply_event(self, event: IEvent) -> None:
    """Apply an FSM event or raise if the transition is invalid."""
    resp = self._fsm.on(event)
    _raise_if_transition_failed(resp.error)
    self.status = self._current_status()

cancel 

cancel() -> None

Cancel a Scheduled or InProgress job and persist Cancelled.

On save() failure the FSM stays Cancelled; retry job.save() only.

Source code in pypepper/scheduler/job.py 
290
291
292
293
294
295
296
297
298
299
def cancel(self) -> None:
    """
    Cancel a Scheduled or InProgress job and persist Cancelled.

    On ``save()`` failure the FSM stays Cancelled; retry ``job.save()`` only.
    """
    self.share_cancel_event()
    self._cancel_event.set()
    self.apply_event(events.CANCEL)
    self.save()

to_record 

to_record() -> JobRecord

Authoritative lifecycle snapshot from the FSM (may lead durable status).

Source code in pypepper/scheduler/job.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
def to_record(self) -> JobRecord:
    """Authoritative lifecycle snapshot from the FSM (may lead durable ``status``)."""
    return JobRecord(
        id=self.id,
        category=self.category,
        channel_id=self.channel_id,
        status=self._current_status(),
        created=self.created,
        updated=self.updated,
        workflow_count=len(self.workflows),
        version=self.version,
        payload=self._workflows_payload(),
    )

from_record classmethod 

from_record(record: JobRecord) -> Job

Rebuild a Job from a store snapshot. Requires registered executor_ids when payload is set.

Leaves the FSM at Unknown (does not restore record.status) so scheduled() can run again.

Source code in pypepper/scheduler/job.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
@classmethod
def from_record(cls, record: JobRecord) -> Job:
    """
    Rebuild a Job from a store snapshot. Requires registered ``executor_id``s when payload is set.

    Leaves the FSM at Unknown (does not restore ``record.status``) so ``scheduled()`` can run again.
    """
    job = cls(category=record.category, channel_id=record.channel_id)
    job.id = record.id
    job.created = record.created
    job.updated = record.updated
    job.version = record.version
    if record.payload:
        job.workflows = cls._workflows_from_payload(record.payload)
    return job

save 

save() -> bool
Source code in pypepper/scheduler/job.py 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
def save(self) -> bool:
    from pypepper.common.config import config as app_config
    from pypepper.scheduler.store.memory import InMemoryJobStore

    app_config.ensure_scheduler_job_store_applied(
        using_default_memory_store=isinstance(get_job_store(), InMemoryJobStore)
    )
    status = self._current_status()
    updated = get_utc_datetime()
    record = JobRecord(
        id=self.id,
        category=self.category,
        channel_id=self.channel_id,
        status=status,
        created=self.created,
        updated=updated,
        workflow_count=len(self.workflows),
        version=self.version,
        payload=self._workflows_payload(),
    )
    store = get_job_store()
    applied = store.put(record)
    if not applied:
        log.debug(f"Job save skipped stale snapshot: id={self.id}, attempted={status}, version={self.version}")
        return False
    durable = store.get(self.id)
    if durable is not None:
        self.version = durable.version
    self.updated = updated
    log.debug(f"Job saved: id={self.id}, channel_id={self.channel_id}, status={status}")
    return True

get_saved staticmethod 

get_saved(job_id: str) -> JobRecord | None
Source code in pypepper/scheduler/job.py 
438
439
440
441
442
443
444
445
446
@staticmethod
def get_saved(job_id: str) -> JobRecord | None:
    from pypepper.common.config import config as app_config
    from pypepper.scheduler.store.memory import InMemoryJobStore

    app_config.ensure_scheduler_job_store_applied(
        using_default_memory_store=isinstance(get_job_store(), InMemoryJobStore)
    )
    return get_job_store().get(job_id)

log 

log() -> None
Source code in pypepper/scheduler/job.py 
448
449
def log(self) -> None:
    log.info(f"Job scheduled: id={self.id}, category={self.category}")

scheduled 

scheduled() -> None
Source code in pypepper/scheduler/job.py 
451
452
def scheduled(self) -> None:
    dispatcher.dispatch(self)

Channel

pypepper.scheduler.channel

Async job channels and channel manager.

SendResult module-attribute 

SendResult = Literal['ok', 'full', 'stopped']

SEND_OK module-attribute 

SEND_OK: SendResult = 'ok'

SEND_FULL module-attribute 

SEND_FULL: SendResult = 'full'

SEND_STOPPED module-attribute 

SEND_STOPPED: SendResult = 'stopped'

manager module-attribute 

manager = ChannelManager()

Channel

Source code in pypepper/scheduler/channel.py 
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
class Channel:
    def __init__(self, maxsize: int = 0) -> None:
        self._stop = False
        self._queue: Queue[Any] = Queue(maxsize)
        # Wakes a blocked ``receive()`` without consuming queue capacity.
        self._stopped = asyncio.Event()
        # Serializes send vs request_stop so a concurrent stop cannot lose an enqueue.
        self._op_lock = Lock()

    @property
    def stop(self) -> bool:
        """True after :meth:`request_stop` (read-only; use ``request_stop()`` to stop)."""
        return self._stop

    async def send(self, value: Any) -> SendResult:
        """
        Enqueue ``value``.

        Returns ``ok``, ``full``, or ``stopped``. The reason is decided under the
        same lock as :meth:`request_stop` (a successful enqueue and a stop-reject
        are mutually exclusive). Prefer ``Job.scheduled()`` for typed errors.
        """
        with self._op_lock:
            if self._stop:
                return SEND_STOPPED
            try:
                self._queue.put_nowait(value)
                return SEND_OK
            except QueueFull:
                return SEND_FULL

    async def receive(self) -> Any | None:
        """
        Wait for the next item, or ``None`` when stop wins and the queue is empty.

        Cases:
        1. ``stop`` already set: non-blocking dequeue via ``get_nowait``, or ``None``
           if empty (deterministic drain).
        2. Live wait: race queue ``get`` against the stop event. A completed ``get``
           is preferred. If stop wins, drain one ready item via ``get_nowait`` so
           leftovers are not abandoned.
        3. ``Worker.run_once`` always calls ``receive`` (including after stop) so
           queued jobs are processed before the consumer exits.

        Do not enqueue ``None`` as a job payload: ``None`` means stop / empty.
        """
        if self.stop:
            try:
                return self._queue.get_nowait()
            except asyncio.QueueEmpty:
                return None

        get_task = asyncio.create_task(self._queue.get())
        stop_task = asyncio.create_task(self._stopped.wait())
        _done, pending = await asyncio.wait(
            {get_task, stop_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in pending:
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await task
        # Prefer a completed get even if it finished while we cancelled siblings.
        if get_task.done() and not get_task.cancelled():
            return get_task.result()
        try:
            return self._queue.get_nowait()
        except asyncio.QueueEmpty:
            return None

    def request_stop(self) -> None:
        """Mark the channel stopped and wake a blocked ``receive()`` if needed."""
        with self._op_lock:
            if self._stop:
                return
            self._stop = True
            self._stopped.set()

    def length(self) -> int:
        return self._queue.qsize()

stop property 

stop: bool

True after :meth:request_stop (read-only; use request_stop() to stop).

send async 

send(value: Any) -> SendResult

Enqueue value.

Returns ok, full, or stopped. The reason is decided under the same lock as :meth:request_stop (a successful enqueue and a stop-reject are mutually exclusive). Prefer Job.scheduled() for typed errors.

Source code in pypepper/scheduler/channel.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
async def send(self, value: Any) -> SendResult:
    """
    Enqueue ``value``.

    Returns ``ok``, ``full``, or ``stopped``. The reason is decided under the
    same lock as :meth:`request_stop` (a successful enqueue and a stop-reject
    are mutually exclusive). Prefer ``Job.scheduled()`` for typed errors.
    """
    with self._op_lock:
        if self._stop:
            return SEND_STOPPED
        try:
            self._queue.put_nowait(value)
            return SEND_OK
        except QueueFull:
            return SEND_FULL

receive async 

receive() -> Any | None

Wait for the next item, or None when stop wins and the queue is empty.

Cases: 1. stop already set: non-blocking dequeue via get_nowait, or None if empty (deterministic drain). 2. Live wait: race queue get against the stop event. A completed get is preferred. If stop wins, drain one ready item via get_nowait so leftovers are not abandoned. 3. Worker.run_once always calls receive (including after stop) so queued jobs are processed before the consumer exits.

Do not enqueue None as a job payload: None means stop / empty.

Source code in pypepper/scheduler/channel.py 
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
async def receive(self) -> Any | None:
    """
    Wait for the next item, or ``None`` when stop wins and the queue is empty.

    Cases:
    1. ``stop`` already set: non-blocking dequeue via ``get_nowait``, or ``None``
       if empty (deterministic drain).
    2. Live wait: race queue ``get`` against the stop event. A completed ``get``
       is preferred. If stop wins, drain one ready item via ``get_nowait`` so
       leftovers are not abandoned.
    3. ``Worker.run_once`` always calls ``receive`` (including after stop) so
       queued jobs are processed before the consumer exits.

    Do not enqueue ``None`` as a job payload: ``None`` means stop / empty.
    """
    if self.stop:
        try:
            return self._queue.get_nowait()
        except asyncio.QueueEmpty:
            return None

    get_task = asyncio.create_task(self._queue.get())
    stop_task = asyncio.create_task(self._stopped.wait())
    _done, pending = await asyncio.wait(
        {get_task, stop_task},
        return_when=asyncio.FIRST_COMPLETED,
    )
    for task in pending:
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task
    # Prefer a completed get even if it finished while we cancelled siblings.
    if get_task.done() and not get_task.cancelled():
        return get_task.result()
    try:
        return self._queue.get_nowait()
    except asyncio.QueueEmpty:
        return None

request_stop 

request_stop() -> None

Mark the channel stopped and wake a blocked receive() if needed.

Source code in pypepper/scheduler/channel.py 
90
91
92
93
94
95
96
def request_stop(self) -> None:
    """Mark the channel stopped and wake a blocked ``receive()`` if needed."""
    with self._op_lock:
        if self._stop:
            return
        self._stop = True
        self._stopped.set()

length 

length() -> int
Source code in pypepper/scheduler/channel.py 
98
99
def length(self) -> int:
    return self._queue.qsize()

ChannelManager

Source code in pypepper/scheduler/channel.py 
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
class ChannelManager:
    _instance: ChannelManager | None = None
    _init_lock = Lock()
    _lock: Lock
    _job_channel: MutableMapping[str, Channel]

    def __new__(cls) -> ChannelManager:
        with cls._init_lock:
            if cls._instance is None:
                inst = super().__new__(cls)
                inst._lock = Lock()
                inst._job_channel = {}
                cls._instance = inst
            return cls._instance

    def __init__(self) -> None:
        pass

    def put(self, key: str, chan: Channel) -> None:
        if not key:
            raise ValueError("invalid key")
        if chan is None:
            raise ValueError("invalid channel")

        with self._lock:
            self._job_channel[key] = chan

    def get(self, key: str) -> Channel | None:
        if not key:
            raise ValueError("invalid key")

        with self._lock:
            if len(self._job_channel) == 0:
                return None

            return self._job_channel.get(key)

    def remove(self, key: str) -> Channel | None:
        if not key:
            raise ValueError("invalid key")

        with self._lock:
            if len(self._job_channel) == 0:
                return None

            return self._job_channel.pop(key)

    def new(self, key: str, maxsize: int = 0) -> Channel:
        """
        Return the channel for ``key``, creating it on first use.

        ``maxsize`` applies only when the channel is created (``0`` = unbounded).
        If the key already exists, the existing channel is returned and ``maxsize``
        is ignored (create bounded channels before Worker/dispatch).
        """
        if not key:
            raise ValueError("invalid key")
        with self._lock:
            chan = self._job_channel.get(key)
            if chan is None:
                chan = Channel(maxsize=maxsize)
                self._job_channel[key] = chan
            elif maxsize != 0 and chan._queue.maxsize != maxsize:
                log.debug(
                    f"Channel {key!r} already exists (maxsize={chan._queue.maxsize}); "
                    f"ignoring requested maxsize={maxsize}"
                )
            return chan

    def available(self, key: str, maxsize: int = 0) -> Channel:
        """Alias for :meth:`new` (``maxsize`` applies only on first create)."""
        return self.new(key, maxsize=maxsize)

put 

put(key: str, chan: Channel) -> None
Source code in pypepper/scheduler/channel.py 
124
125
126
127
128
129
130
131
def put(self, key: str, chan: Channel) -> None:
    if not key:
        raise ValueError("invalid key")
    if chan is None:
        raise ValueError("invalid channel")

    with self._lock:
        self._job_channel[key] = chan

get 

get(key: str) -> Channel | None
Source code in pypepper/scheduler/channel.py 
133
134
135
136
137
138
139
140
141
def get(self, key: str) -> Channel | None:
    if not key:
        raise ValueError("invalid key")

    with self._lock:
        if len(self._job_channel) == 0:
            return None

        return self._job_channel.get(key)

remove 

remove(key: str) -> Channel | None
Source code in pypepper/scheduler/channel.py 
143
144
145
146
147
148
149
150
151
def remove(self, key: str) -> Channel | None:
    if not key:
        raise ValueError("invalid key")

    with self._lock:
        if len(self._job_channel) == 0:
            return None

        return self._job_channel.pop(key)

new 

new(key: str, maxsize: int = 0) -> Channel

Return the channel for key, creating it on first use.

maxsize applies only when the channel is created (0 = unbounded). If the key already exists, the existing channel is returned and maxsize is ignored (create bounded channels before Worker/dispatch).

Source code in pypepper/scheduler/channel.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def new(self, key: str, maxsize: int = 0) -> Channel:
    """
    Return the channel for ``key``, creating it on first use.

    ``maxsize`` applies only when the channel is created (``0`` = unbounded).
    If the key already exists, the existing channel is returned and ``maxsize``
    is ignored (create bounded channels before Worker/dispatch).
    """
    if not key:
        raise ValueError("invalid key")
    with self._lock:
        chan = self._job_channel.get(key)
        if chan is None:
            chan = Channel(maxsize=maxsize)
            self._job_channel[key] = chan
        elif maxsize != 0 and chan._queue.maxsize != maxsize:
            log.debug(
                f"Channel {key!r} already exists (maxsize={chan._queue.maxsize}); "
                f"ignoring requested maxsize={maxsize}"
            )
        return chan

available 

available(key: str, maxsize: int = 0) -> Channel

Alias for :meth:new (maxsize applies only on first create).

Source code in pypepper/scheduler/channel.py 
175
176
177
def available(self, key: str, maxsize: int = 0) -> Channel:
    """Alias for :meth:`new` (``maxsize`` applies only on first create)."""
    return self.new(key, maxsize=maxsize)

new 

new(maxsize: int = 0) -> Channel
Source code in pypepper/scheduler/channel.py 
102
103
def new(maxsize: int = 0) -> Channel:
    return Channel(maxsize=maxsize)

Worker

pypepper.scheduler.worker

Channel consumer that runs job workflows.

Worker

Consume jobs from a channel and run their workflows.

Source code in pypepper/scheduler/worker.py 
 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
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
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
class Worker:
    """Consume jobs from a channel and run their workflows."""

    def __init__(self, channel: Channel) -> None:
        self.channel = channel

    async def run_once(self) -> Job | None:
        raw = await self.channel.receive()
        if raw is None:
            return None
        job = cast(Job, raw)
        await self._process(job)
        return job

    async def run_forever(self) -> None:
        stopped_redelivery: JobRedeliveryError | None = None
        while True:
            try:
                job = await self.run_once()
            except JobRequeuedError as e:
                # Job is back on the channel; re-raise so supervisors see failure
                # (do not continue into a persist-failure busy-spin).
                log.error(f"Worker run_forever stopping after re-enqueue: {e!r}")
                raise
            except JobRedeliveryError as e:
                # Full: stop immediately. Stopped: finish draining leftovers, then
                # re-raise outside this handler so supervisors still see non-success
                # (the unrestored job is not on the channel).
                if e.reason == "stopped":
                    log.error(
                        f"Worker run_forever drain continues after redelivery "
                        f"(channel stopped; job not on channel): {e!r}"
                    )
                    if stopped_redelivery is None:
                        stopped_redelivery = e
                    continue
                raise
            except Exception as e:
                # Intentional behavior change vs raise-and-exit: log and continue.
                # Continue-on-error does not redeliver by itself; RUN-start restore
                # paths re-enqueue inside ``_process`` when possible.
                log.error(f"Worker run_forever job error (continuing): {e!r}")
                continue

            if job is None:
                if stopped_redelivery is not None:
                    raise stopped_redelivery
                return

    async def _requeue_after_run_restore(self, job: Job, save_exc: BaseException) -> NoReturn:
        """Re-enqueue after pre-RUN restore, or persist Failed and raise. Never returns normally."""
        result = await self.channel.send(job)
        if result == SEND_OK:
            log.error(f"Job re-enqueued after RUN persist restore: id={job.id}")
            raise JobRequeuedError(
                f"Job re-enqueued after RUN persist restore: id={job.id}, channel_id={job.channel_id}"
            ) from save_exc
        reason = "full" if result == SEND_FULL else "stopped"
        _mark_undeliverable_failed(job)
        raise JobRedeliveryError(
            f"Job RUN persist restore could not re-enqueue "
            f"(channel {reason}): id={job.id}, channel_id={job.channel_id}",
            reason=reason,
            job=job,
        ) from save_exc

    async def _fail_or_restore_after_run_persist(
        self,
        job: Job,
        prev_state: IState | None,
        prev_status: str,
        save_exc: BaseException,
    ) -> None:
        try:
            if job.is_cancelled():
                raise RuntimeError(f"Job already cancelled: id={job.id}")
            _persist_failed_snapshot(job)
        except Exception as fail_save_exc:
            if job.is_cancelled():
                try:
                    _ensure_cancelled_persisted(job)
                except Exception as cancel_save_exc:
                    log.error(
                        f"Job RUN persist failed: id={job.id}, error={save_exc}; "
                        f"cancel won but Cancelled persist failed: {cancel_save_exc}"
                    )
                    raise cancel_save_exc from save_exc
                log.error(
                    f"Job RUN persist failed: id={job.id}, error={save_exc}; "
                    f"cancel already applied (skip restore): {fail_save_exc}"
                )
                raise save_exc from fail_save_exc
            job.restore_lifecycle(prev_state, prev_status)
            log.error(
                f"Job RUN persist failed: id={job.id}, error={save_exc}; FAIL persist also failed: {fail_save_exc}"
            )
            await self._requeue_after_run_restore(job, save_exc)
        log.error(
            f"Job RUN persist failed: id={job.id}, error={save_exc}; persisted Failed instead (do not re-run workflows)"
        )
        raise save_exc

    async def _process(self, job: Job) -> None:
        if job.is_cancelled():
            log.info(f"Job already cancelled, skip: id={job.id}")
            _ensure_cancelled_persisted_logged(job)
            return

        prev_state = job._fsm.current()
        prev_status = job.status

        try:
            if Job.get_saved(job.id) is None and not job.save() and Job.get_saved(job.id) is None:
                raise RuntimeError(f"Job Scheduled persist failed before RUN: id={job.id}")
        except Exception as heal_exc:
            log.error(f"Job missing Scheduled snapshot before RUN: id={job.id}, error={heal_exc}")
            await self._fail_or_restore_after_run_persist(job, prev_state, prev_status, heal_exc)
            return

        job.apply_event(events.RUN)
        try:
            applied = job.save()
        except Exception as save_exc:
            await self._fail_or_restore_after_run_persist(job, prev_state, prev_status, save_exc)
            return

        if not applied:
            durable = Job.get_saved(job.id)
            durable_status = None if durable is None else durable.status
            if durable_status == Status.CANCELLED.value:
                if not job.is_cancelled():
                    job.apply_event(events.CANCEL)
                _ensure_cancelled_persisted_logged(job)
                return
            if durable_status in _FOLLOW_WITHOUT_RUN:
                _follow_durable_status(job, durable_status)
                log.info(f"Job RUN snapshot skipped; following durable {durable_status}: id={job.id}")
                return
            await self._fail_or_restore_after_run_persist(
                job,
                prev_state,
                prev_status,
                RuntimeError(f"Job RUN persist skipped: id={job.id}, durable={durable_status}"),
            )
            return

        try:
            workflows = getattr(job, "workflows", None) or []
            job.share_cancel_event()
            for workflow in workflows:
                if job.is_cancelled():
                    log.info(f"Job cancelled between workflows: id={job.id}")
                    _ensure_cancelled_persisted_logged(job)
                    return
                # Workflow.run is sync; run it in a worker thread.
                await asyncio.to_thread(workflow.run)
        except Exception as e:
            if job.is_cancelled():
                log.info(f"Job cancelled (workflow error ignored): id={job.id}, error={e}")
                _ensure_cancelled_persisted_logged(job, cause=e)
                return
            try:
                _transition_and_save_terminal(job, events.FAIL)
            except Exception as save_exc:
                log.error(
                    f"Job failed: id={job.id}, error={e}; "
                    f"terminal persist also failed (retry job.save only): {save_exc}"
                )
                raise e from save_exc
            log.error(f"Job failed: id={job.id}, error={e}")
            raise

        if job.is_cancelled():
            log.info(f"Job cancelled before complete: id={job.id}")
            _ensure_cancelled_persisted_logged(job)
            return

        try:
            _transition_and_save_terminal(job, events.COMPLETE)
        except Exception as save_exc:
            log.error(
                f"Job completed but terminal persist failed: id={job.id}; "
                f"retry job.save only (do not re-run workflows): {save_exc}"
            )
            raise
        log.info(f"Job completed: id={job.id}")

channel instance-attribute 

channel = channel

run_once async 

run_once() -> Job | None
Source code in pypepper/scheduler/worker.py 
 98
 99
100
101
102
103
104
async def run_once(self) -> Job | None:
    raw = await self.channel.receive()
    if raw is None:
        return None
    job = cast(Job, raw)
    await self._process(job)
    return job

run_forever async 

run_forever() -> None
Source code in pypepper/scheduler/worker.py 
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
async def run_forever(self) -> None:
    stopped_redelivery: JobRedeliveryError | None = None
    while True:
        try:
            job = await self.run_once()
        except JobRequeuedError as e:
            # Job is back on the channel; re-raise so supervisors see failure
            # (do not continue into a persist-failure busy-spin).
            log.error(f"Worker run_forever stopping after re-enqueue: {e!r}")
            raise
        except JobRedeliveryError as e:
            # Full: stop immediately. Stopped: finish draining leftovers, then
            # re-raise outside this handler so supervisors still see non-success
            # (the unrestored job is not on the channel).
            if e.reason == "stopped":
                log.error(
                    f"Worker run_forever drain continues after redelivery "
                    f"(channel stopped; job not on channel): {e!r}"
                )
                if stopped_redelivery is None:
                    stopped_redelivery = e
                continue
            raise
        except Exception as e:
            # Intentional behavior change vs raise-and-exit: log and continue.
            # Continue-on-error does not redeliver by itself; RUN-start restore
            # paths re-enqueue inside ``_process`` when possible.
            log.error(f"Worker run_forever job error (continuing): {e!r}")
            continue

        if job is None:
            if stopped_redelivery is not None:
                raise stopped_redelivery
            return

Executor

pypepper.scheduler.executor

Task executor interfaces, callable adapter, and named registry.

ExecutorFactory module-attribute 

ExecutorFactory = Callable[[], 'IExecutor']

executor_registry module-attribute 

executor_registry = ExecutorRegistry()

IExecutor

Source code in pypepper/scheduler/executor.py 
17
18
19
20
class IExecutor(metaclass=ABCMeta):
    @abstractmethod
    def execute(self, task: Task, context: Context | None = None) -> Any:
        pass

execute abstractmethod 

execute(task: Task, context: Context | None = None) -> Any
Source code in pypepper/scheduler/executor.py 
18
19
20
@abstractmethod
def execute(self, task: Task, context: Context | None = None) -> Any:
    pass

Executor

Bases: IExecutor

No-op executor (placeholder for tasks without work).

Source code in pypepper/scheduler/executor.py 
23
24
25
26
27
class Executor(IExecutor):
    """No-op executor (placeholder for tasks without work)."""

    def execute(self, task: Task, context: Context | None = None) -> Any:
        return None

execute 

execute(task: Task, context: Context | None = None) -> Any
Source code in pypepper/scheduler/executor.py 
26
27
def execute(self, task: Task, context: Context | None = None) -> Any:
    return None

CallableExecutor

Bases: Executor

Executor that runs a provided callable.

Source code in pypepper/scheduler/executor.py 
30
31
32
33
34
35
36
37
class CallableExecutor(Executor):
    """Executor that runs a provided callable."""

    def __init__(self, func: Callable[..., Any]) -> None:
        self._func = func

    def execute(self, task: Task, context: Context | None = None) -> Any:
        return self._func(task, context)

execute 

execute(task: Task, context: Context | None = None) -> Any
Source code in pypepper/scheduler/executor.py 
36
37
def execute(self, task: Task, context: Context | None = None) -> Any:
    return self._func(task, context)

ExecutorRegistry

Explicit process-wide registry of named executor factories.

Source code in pypepper/scheduler/executor.py 
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
class ExecutorRegistry:
    """Explicit process-wide registry of named executor factories."""

    _instance: ExecutorRegistry | None = None
    _init_lock = Lock()
    _lock: Lock
    _factories: dict[str, ExecutorFactory | IExecutor]

    def __new__(cls) -> ExecutorRegistry:
        with cls._init_lock:
            if cls._instance is None:
                inst = super().__new__(cls)
                inst._lock = Lock()
                inst._factories = {}
                cls._instance = inst
            return cls._instance

    def register(self, name: str, factory: ExecutorFactory | IExecutor) -> None:
        if not name:
            raise ValueError("invalid executor id")
        if factory is None:
            raise ValueError("invalid executor factory")
        with self._lock:
            self._factories[name] = factory

    def resolve(self, name: str) -> IExecutor:
        if not name:
            raise ValueError("invalid executor id")
        with self._lock:
            spec = self._factories.get(name)
        if spec is None:
            raise ValueError(f"unknown executor_id: {name}")
        if isinstance(spec, IExecutor):
            return spec
        return spec()

    def clear(self) -> None:
        with self._lock:
            self._factories.clear()

register 

register(
    name: str, factory: ExecutorFactory | IExecutor
) -> None
Source code in pypepper/scheduler/executor.py 
57
58
59
60
61
62
63
def register(self, name: str, factory: ExecutorFactory | IExecutor) -> None:
    if not name:
        raise ValueError("invalid executor id")
    if factory is None:
        raise ValueError("invalid executor factory")
    with self._lock:
        self._factories[name] = factory

resolve 

resolve(name: str) -> IExecutor
Source code in pypepper/scheduler/executor.py 
65
66
67
68
69
70
71
72
73
74
def resolve(self, name: str) -> IExecutor:
    if not name:
        raise ValueError("invalid executor id")
    with self._lock:
        spec = self._factories.get(name)
    if spec is None:
        raise ValueError(f"unknown executor_id: {name}")
    if isinstance(spec, IExecutor):
        return spec
    return spec()

clear 

clear() -> None
Source code in pypepper/scheduler/executor.py 
76
77
78
def clear(self) -> None:
    with self._lock:
        self._factories.clear()

Events

pypepper.scheduler.events

Scheduler FSM events and builder.

INIT module-attribute 

INIT = event.new(name='init', src=Status.UNKNOWN)

SCHEDULE module-attribute 

SCHEDULE = event.new(
    name="schedule", src=Status.INITIALIZING
)

RUN module-attribute 

RUN = event.new(name='run', src=Status.SCHEDULED)

FAIL module-attribute 

FAIL = event.new(name='fail', src=Status.IN_PROGRESS)

COMPLETE module-attribute 

COMPLETE = event.new(
    name="complete", src=Status.IN_PROGRESS
)

CANCEL module-attribute 

CANCEL = event.new(name='cancel', src=Status.IN_PROGRESS)

FSM

Per-instance scheduler FSM wrapper (backward-compatible API).

Source code in pypepper/scheduler/events.py 
61
62
63
64
65
66
67
68
class FSM:
    """Per-instance scheduler FSM wrapper (backward-compatible API)."""

    def __init__(self) -> None:
        self._machine = build_scheduler_fsm()

    def on(self, evt: Event) -> IResponse:
        return self._machine.on(evt)

on 

on(evt: Event) -> IResponse
Source code in pypepper/scheduler/events.py 
67
68
def on(self, evt: Event) -> IResponse:
    return self._machine.on(evt)

build_scheduler_fsm 

build_scheduler_fsm() -> fsm.FSM

Create a new scheduler FSM instance (per job).

Source code in pypepper/scheduler/events.py 
56
57
58
def build_scheduler_fsm() -> fsm.FSM:
    """Create a new scheduler FSM instance (per job)."""
    return fsm.new(_OPTIONS)

Job store

pypepper.scheduler.store

Pluggable job persistence (memory / postgres / mysql / mongodb).

Backend module-attribute 

Backend = Literal['memory', 'postgres', 'mysql', 'mongodb']

job_store module-attribute 

job_store: IJobStore = _job_store

IJobStore

Pluggable persistence for JobRecord snapshots.

Source code in pypepper/scheduler/store/interfaces.py 
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
class IJobStore(metaclass=ABCMeta):
    """Pluggable persistence for JobRecord snapshots."""

    @abstractmethod
    def put(self, record: JobRecord) -> bool:
        """
        Upsert by ``id``.

        Returns ``True`` when the row is inserted or updated, ``False`` when the
        write is skipped (lifecycle fence or version conflict).

        Must not overwrite an existing row's ``created``. Must not replace a
        durable status with an earlier lifecycle (for example Scheduled must
        not overwrite InProgress/Completed/Failed/Cancelled). Distinct
        terminals must not overwrite each other. Same status may update other
        fields. Updates require ``record.version`` to equal the durable version
        (then the stored version is incremented).
        """
        pass

    @abstractmethod
    def get(self, job_id: str) -> JobRecord | None:
        pass

    @abstractmethod
    def delete(self, job_id: str) -> None:
        pass

    @abstractmethod
    def list(self, channel_id: str | None = None) -> list[JobRecord]:
        pass

    @abstractmethod
    def clear(self) -> None:
        pass

put abstractmethod 

put(record: JobRecord) -> bool

Upsert by id.

Returns True when the row is inserted or updated, False when the write is skipped (lifecycle fence or version conflict).

Must not overwrite an existing row's created. Must not replace a durable status with an earlier lifecycle (for example Scheduled must not overwrite InProgress/Completed/Failed/Cancelled). Distinct terminals must not overwrite each other. Same status may update other fields. Updates require record.version to equal the durable version (then the stored version is incremented).

Source code in pypepper/scheduler/store/interfaces.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@abstractmethod
def put(self, record: JobRecord) -> bool:
    """
    Upsert by ``id``.

    Returns ``True`` when the row is inserted or updated, ``False`` when the
    write is skipped (lifecycle fence or version conflict).

    Must not overwrite an existing row's ``created``. Must not replace a
    durable status with an earlier lifecycle (for example Scheduled must
    not overwrite InProgress/Completed/Failed/Cancelled). Distinct
    terminals must not overwrite each other. Same status may update other
    fields. Updates require ``record.version`` to equal the durable version
    (then the stored version is incremented).
    """
    pass

get abstractmethod 

get(job_id: str) -> JobRecord | None
Source code in pypepper/scheduler/store/interfaces.py 
45
46
47
@abstractmethod
def get(self, job_id: str) -> JobRecord | None:
    pass

delete abstractmethod 

delete(job_id: str) -> None
Source code in pypepper/scheduler/store/interfaces.py 
49
50
51
@abstractmethod
def delete(self, job_id: str) -> None:
    pass

list abstractmethod 

list(channel_id: str | None = None) -> list[JobRecord]
Source code in pypepper/scheduler/store/interfaces.py 
53
54
55
@abstractmethod
def list(self, channel_id: str | None = None) -> list[JobRecord]:
    pass

clear abstractmethod 

clear() -> None
Source code in pypepper/scheduler/store/interfaces.py 
57
58
59
@abstractmethod
def clear(self) -> None:
    pass

JobRecord dataclass

Job snapshot. Executors are omitted unless payload is present.

Source code in pypepper/scheduler/store/interfaces.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
@dataclass(frozen=True)
class JobRecord:
    """Job snapshot. Executors are omitted unless ``payload`` is present."""

    id: str
    category: str | None
    channel_id: str
    status: str
    created: str
    updated: str
    workflow_count: int = 0
    version: int = 1
    payload: dict[str, Any] | None = None

id instance-attribute 

id: str

category instance-attribute 

category: str | None

channel_id instance-attribute 

channel_id: str

status instance-attribute 

status: str

created instance-attribute 

created: str

updated instance-attribute 

updated: str

workflow_count class-attribute instance-attribute 

workflow_count: int = 0

version class-attribute instance-attribute 

version: int = 1

payload class-attribute instance-attribute 

payload: dict[str, Any] | None = None

InMemoryJobStore

Bases: IJobStore

Thread-safe dict-backed job store.

Source code in pypepper/scheduler/store/memory.py 
12
13
14
15
16
17
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
class InMemoryJobStore(IJobStore):
    """Thread-safe dict-backed job store."""

    def __init__(self) -> None:
        self._lock = Lock()
        self._store: dict[str, JobRecord] = {}

    def put(self, record: JobRecord) -> bool:
        with self._lock:
            existing = self._store.get(record.id)
            if existing is None:
                self._store[record.id] = record
                return True
            if not put_may_write(existing.status, existing.version, record.status, record.version):
                return False
            self._store[record.id] = replace(
                record,
                created=existing.created,
                version=existing.version + 1,
            )
            return True

    def get(self, job_id: str) -> JobRecord | None:
        with self._lock:
            return self._store.get(job_id)

    def delete(self, job_id: str) -> None:
        with self._lock:
            self._store.pop(job_id, None)

    def list(self, channel_id: str | None = None) -> list[JobRecord]:
        with self._lock:
            records = list(self._store.values())
        if channel_id is None:
            return records
        return [r for r in records if r.channel_id == channel_id]

    def clear(self) -> None:
        with self._lock:
            self._store.clear()

put 

put(record: JobRecord) -> bool
Source code in pypepper/scheduler/store/memory.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def put(self, record: JobRecord) -> bool:
    with self._lock:
        existing = self._store.get(record.id)
        if existing is None:
            self._store[record.id] = record
            return True
        if not put_may_write(existing.status, existing.version, record.status, record.version):
            return False
        self._store[record.id] = replace(
            record,
            created=existing.created,
            version=existing.version + 1,
        )
        return True

get 

get(job_id: str) -> JobRecord | None
Source code in pypepper/scheduler/store/memory.py 
34
35
36
def get(self, job_id: str) -> JobRecord | None:
    with self._lock:
        return self._store.get(job_id)

delete 

delete(job_id: str) -> None
Source code in pypepper/scheduler/store/memory.py 
38
39
40
def delete(self, job_id: str) -> None:
    with self._lock:
        self._store.pop(job_id, None)

list 

list(channel_id: str | None = None) -> list[JobRecord]
Source code in pypepper/scheduler/store/memory.py 
42
43
44
45
46
47
def list(self, channel_id: str | None = None) -> list[JobRecord]:
    with self._lock:
        records = list(self._store.values())
    if channel_id is None:
        return records
    return [r for r in records if r.channel_id == channel_id]

clear 

clear() -> None
Source code in pypepper/scheduler/store/memory.py 
49
50
51
def clear(self) -> None:
    with self._lock:
        self._store.clear()

get_job_store 

get_job_store() -> IJobStore
Source code in pypepper/scheduler/store/__init__.py 
20
21
def get_job_store() -> IJobStore:
    return _job_store

reset_job_store_mismatch_warning 

reset_job_store_mismatch_warning() -> None

Reset the durable-vs-memory mismatch one-shot warn (tests).

Source code in pypepper/scheduler/store/__init__.py 
61
62
63
64
65
def reset_job_store_mismatch_warning() -> None:
    """Reset the durable-vs-memory mismatch one-shot warn (tests)."""
    global _mismatch_warned
    with _mismatch_warn_lock:
        _mismatch_warned = False

set_job_store 

set_job_store(store: IJobStore) -> None

Replace the process-wide job store and clear any deferred durable YAML flag.

Source code in pypepper/scheduler/store/__init__.py 
74
75
76
77
78
79
80
81
82
83
def set_job_store(store: IJobStore) -> None:
    """Replace the process-wide job store and clear any deferred durable YAML flag."""
    global _job_store, job_store
    if isinstance(store, InMemoryJobStore):
        declared = _yaml_declared_job_store_backend()
        if declared is not None and declared.strip().lower() not in ("", "memory"):
            _warn_durable_vs_memory_once(declared)
    _job_store = store
    job_store = store
    _mark_job_store_applied()

reset_job_store 

reset_job_store() -> None

Reset to a fresh in-memory store (tests / process reset).

Does not acknowledge a deferred durable YAML backend: if the current config still declares a non-memory jobStore, deferred fail-fast is re-armed so Job.save cannot silently use memory.

Also clears the durable-vs-memory mismatch one-shot so a later explicit memory install can warn again for the new cycle.

Source code in pypepper/scheduler/store/__init__.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def reset_job_store() -> None:
    """
    Reset to a fresh in-memory store (tests / process reset).

    Does **not** acknowledge a deferred durable YAML backend: if the current
    config still declares a non-memory ``jobStore``, deferred fail-fast is
    re-armed so ``Job.save`` cannot silently use memory.

    Also clears the durable-vs-memory mismatch one-shot so a later explicit
    memory install can warn again for the new cycle.
    """
    global _job_store, job_store
    _job_store = InMemoryJobStore()
    job_store = _job_store
    reset_job_store_mismatch_warning()
    from pypepper.common.config import config as app_config

    app_config.refresh_scheduler_job_store_deferred()

configure_job_store 

configure_job_store(
    backend: Backend = "memory", **kwargs: Any
) -> IJobStore

Build and install a job store.

Parameters:

Name Type Description Default
backend Backend

memory | postgres | mysql | mongodb

 'memory'
**kwargs Any

Connection options forwarded to the backend (e.g. uri, host, port, username, password, db).

 {}
Source code in pypepper/scheduler/store/__init__.py 
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
def configure_job_store(backend: Backend = "memory", **kwargs: Any) -> IJobStore:
    """
    Build and install a job store.

    Parameters
    ----------
    backend:
        ``memory`` | ``postgres`` | ``mysql`` | ``mongodb``
    **kwargs:
        Connection options forwarded to the backend (e.g. ``uri``, ``host``,
        ``port``, ``username``, ``password``, ``db``).
    """
    if backend == "memory":
        store: IJobStore = InMemoryJobStore()
    elif backend in ("postgres", "mysql"):
        from pypepper.scheduler.store.sql import SqlJobStore

        store = SqlJobStore(backend=backend, **kwargs)
    elif backend == "mongodb":
        from pypepper.scheduler.store.mongodb import MongoJobStore

        store = MongoJobStore(**kwargs)
    else:
        raise ValueError(f"unsupported job store backend: {backend!r}")

    set_job_store(store)
    return store

setup_from_config 

setup_from_config(yml_config: Any | None = None) -> None

Configure job store from YAML scheduler.jobStore (optional).

Source code in pypepper/scheduler/store/__init__.py 
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
def setup_from_config(yml_config: Any | None = None) -> None:
    """Configure job store from YAML ``scheduler.jobStore`` (optional)."""
    if yml_config is None or not hasattr(yml_config, "scheduler"):
        return
    scheduler_cfg = yml_config.scheduler
    if scheduler_cfg is None or not hasattr(scheduler_cfg, "jobStore"):
        return
    store_cfg = scheduler_cfg.jobStore
    if store_cfg is None:
        return

    backend_raw = getattr(store_cfg, "backend", None) or "memory"
    if backend_raw not in _VALID_BACKENDS:
        raise ValueError(f"unsupported job store backend: {backend_raw!r}")
    backend = cast(Backend, backend_raw)
    # Avoid wiping an existing in-memory store when config still says memory.
    if backend == "memory" and isinstance(get_job_store(), InMemoryJobStore):
        _mark_job_store_applied()
        return

    kwargs: dict[str, Any] = {}
    for key in (
        "uri",
        "host",
        "port",
        "username",
        "password",
        "db",
        "sslmode",
        "charset",
        "auth_source",
    ):
        if hasattr(store_cfg, key):
            value = getattr(store_cfg, key)
            if value is not None:
                kwargs[key] = value

    configure_job_store(backend, **kwargs)