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: Executor

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: Executor

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
class Task(ITask):
    def __init__(
        self,
        channel_id: str,
        dag_id: str,
        fingerprint: str,
        name: str,
        category: str,
        description: str,
        tags: list[Tag],
        executor: Executor,
        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,
    ) -> 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 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
        # Seconds per execute attempt; 0 = no timeout. Soft orphan/overlap only when > 0.
        self.round_timeout = round_timeout
        # 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

round_timeout instance-attribute 

round_timeout = round_timeout

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 
60
61
class IWorkflow(IBase, metaclass=ABCMeta):
    tasks: list[Task]

tasks instance-attribute 

tasks: list[Task]

Workflow

Bases: IWorkflow

Source code in pypepper/scheduler/workflow.py 
 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
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 soft-timeout per execute attempt (0 = none). Timed-out
          work that already started may keep running on the shared soft-timeout pool; the
          next attempt can overlap. Concurrent soft-timeout executes are capped
          (``_SOFT_TIMEOUT_MAX_WORKERS``, including orphans); further work queues and a
          short timeout may fire before the task starts. Queued work that times out
          before start is cancelled when possible so it does not run later.
        - 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))

        # Soft timeout via shared pool: do not shut down the pool so TimeoutError
        # fails fast and retries can proceed while started orphaned work may still run.
        future: Future[object | None] = _soft_timeout_pool().submit(executor.execute, task, task.context)
        try:
            return cast(object | None, future.result(timeout=timeout))
        except FuturesTimeoutError as e:
            # On 3.10+, FuturesTimeoutError is TimeoutError. If the pool Future finished
            # in the race window, return/raise its outcome; otherwise wrap the wait timeout.
            if future.done():
                return cast(object | None, future.result())
            # Prefer cancelling queued work so a "failed" attempt does not run later.
            if future.cancel():
                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())
            future.add_done_callback(partial(_log_soft_timeout_orphan, task_id=task.id, task_name=task.name))
            raise TimeoutError(
                f"Task execute exceeded round_timeout={timeout}s "
                f"(execute still running): id={task.id}, name={task.name}"
            ) from e

    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 
68
69
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 
71
72
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 
74
75
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 soft-timeout per execute attempt (0 = none). Timed-out work that already started may keep running on the shared soft-timeout pool; the next attempt can overlap. Concurrent soft-timeout executes are capped (_SOFT_TIMEOUT_MAX_WORKERS, including orphans); further work queues and a short timeout may fire before the task starts. Queued work that times out before start is cancelled when possible so it does not run later. - 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 
 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
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 soft-timeout per execute attempt (0 = none). Timed-out
      work that already started may keep running on the shared soft-timeout pool; the
      next attempt can overlap. Concurrent soft-timeout executes are capped
      (``_SOFT_TIMEOUT_MAX_WORKERS``, including orphans); further work queues and a
      short timeout may fire before the task starts. Queued work that times out
      before start is cancelled when possible so it does not run later.
    - 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.

dispatcher module-attribute 

dispatcher = Dispatcher()

ChannelFullError

Bases: RuntimeError

Bounded channel rejected an enqueue (pre-execution; safe to roll back).

Source code in pypepper/scheduler/job.py 
24
25
class ChannelFullError(RuntimeError):
    """Bounded channel rejected an enqueue (pre-execution; safe to roll back)."""

Processor

Source code in pypepper/scheduler/job.py 
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
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, call "
            "job.save(), then await Channel.send(job) and consume with Worker"
        )

    @staticmethod
    async def async_run(
        job: Job,
        chan: Channel,
        *,
        on_enqueued: Callable[[], None] | None = None,
    ) -> None:
        ok = await chan.send(job)
        if not 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 
37
38
39
40
41
42
43
44
45
46
47
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, call "
        "job.save(), then await Channel.send(job) 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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@staticmethod
async def async_run(
    job: Job,
    chan: Channel,
    *,
    on_enqueued: Callable[[], None] | None = None,
) -> None:
    ok = await chan.send(job)
    if not 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 
 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
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:
        assert key, "invalid key"
        assert processor, "invalid processor"

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

    def _get_processor(self, key: str) -> Processor | None:
        assert key, "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 schedule/enqueue failure must roll back so retry can re-enter.
        prev_state = job._fsm.current()
        prev_status = job.status
        try:
            job.apply_event(events.INIT)
            job.apply_event(events.SCHEDULE)
            job.save()
        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, exceptions are committed-enqueue + secondary failure.
        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)
        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)
            try:
                get_job_store().delete(job.id)
            except Exception as delete_exc:
                log.error(f"Job enqueue cleanup delete failed: id={job.id}, error={delete_exc}")
            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 
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
def dispatch(self, job: Job) -> None:
    # Pre-channel schedule/enqueue failure must roll back so retry can re-enter.
    prev_state = job._fsm.current()
    prev_status = job.status
    try:
        job.apply_event(events.INIT)
        job.apply_event(events.SCHEDULE)
        job.save()
    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, exceptions are committed-enqueue + secondary failure.
    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)
    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)
        try:
            get_job_store().delete(job.id)
        except Exception as delete_exc:
            log.error(f"Job enqueue cleanup delete failed: id={job.id}, error={delete_exc}")
        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 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class IJob(IBase, metaclass=ABCMeta):
    workflows: list[Workflow]

    @abstractmethod
    def save(self) -> None:
        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() -> None
Source code in pypepper/scheduler/job.py 
159
160
161
@abstractmethod
def save(self) -> None:
    pass

log abstractmethod 

log() -> None
Source code in pypepper/scheduler/job.py 
163
164
165
@abstractmethod
def log(self) -> None:
    pass

scheduled abstractmethod 

scheduled() -> None
Source code in pypepper/scheduler/job.py 
167
168
169
@abstractmethod
def scheduled(self) -> None:
    pass

cancel abstractmethod 

cancel() -> None
Source code in pypepper/scheduler/job.py 
171
172
173
@abstractmethod
def cancel(self) -> None:
    pass

is_cancelled abstractmethod 

is_cancelled() -> bool
Source code in pypepper/scheduler/job.py 
175
176
177
@abstractmethod
def is_cancelled(self) -> bool:
    pass

Job

Bases: IJob

Source code in pypepper/scheduler/job.py 
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 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

    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)

    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.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,
        )

    def save(self) -> 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)
        )
        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,
        )
        get_job_store().put(record)
        # Mutate in-memory fields only after durable persist succeeds.
        self.status = status
        self.updated = updated
        log.debug(f"Job saved: id={self.id}, channel_id={self.channel_id}, status={self.status}")

    @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

is_cancelled 

is_cancelled() -> bool
Source code in pypepper/scheduler/job.py 
203
204
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 
206
207
208
209
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 
211
212
213
214
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)

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 
216
217
218
219
220
221
222
223
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.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 
225
226
227
228
229
230
231
232
233
234
235
236
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,
    )

save 

save() -> None
Source code in pypepper/scheduler/job.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def save(self) -> 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)
    )
    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,
    )
    get_job_store().put(record)
    # Mutate in-memory fields only after durable persist succeeds.
    self.status = status
    self.updated = updated
    log.debug(f"Job saved: id={self.id}, channel_id={self.channel_id}, status={self.status}")

get_saved staticmethod 

get_saved(job_id: str) -> JobRecord | None
Source code in pypepper/scheduler/job.py 
263
264
265
266
267
268
269
270
271
@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 
273
274
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 
276
277
def scheduled(self) -> None:
    dispatcher.dispatch(self)

Channel

pypepper.scheduler.channel

Async job channels and channel manager.

manager module-attribute 

manager = ChannelManager()

Channel

Source code in pypepper/scheduler/channel.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class Channel:
    def __init__(self, maxsize: int = 0):
        self.stop = False
        self._queue: Queue[Any] = Queue(maxsize)

    async def send(self, value: Any) -> bool:
        try:
            self._queue.put_nowait(value)
            return True
        except QueueFull:
            return False

    async def receive(self):
        return await self._queue.get()

    def length(self):
        return self._queue.qsize()

stop instance-attribute 

stop = False

send async 

send(value: Any) -> bool
Source code in pypepper/scheduler/channel.py 
18
19
20
21
22
23
async def send(self, value: Any) -> bool:
    try:
        self._queue.put_nowait(value)
        return True
    except QueueFull:
        return False

receive async 

receive()
Source code in pypepper/scheduler/channel.py 
25
26
async def receive(self):
    return await self._queue.get()

length 

length()
Source code in pypepper/scheduler/channel.py 
28
29
def length(self):
    return self._queue.qsize()

ChannelManager

Source code in pypepper/scheduler/channel.py 
 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
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:
        assert key, "invalid key"
        assert chan, "invalid channel"

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

    def get(self, key: str) -> Channel | None:
        assert key, "invalid key"

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

            return self._job_channel.get(key)

    def remove(self, key: str):
        assert key, "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).
        """
        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 
54
55
56
57
58
59
def put(self, key: str, chan: Channel) -> None:
    assert key, "invalid key"
    assert chan, "invalid channel"

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

get 

get(key: str) -> Channel | None
Source code in pypepper/scheduler/channel.py 
61
62
63
64
65
66
67
68
def get(self, key: str) -> Channel | None:
    assert key, "invalid key"

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

        return self._job_channel.get(key)

remove 

remove(key: str)
Source code in pypepper/scheduler/channel.py 
70
71
72
73
74
75
76
77
def remove(self, key: str):
    assert key, "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 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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).
    """
    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 
 99
100
101
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 
32
33
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 
 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
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:
        if self.channel.stop:
            return None

        job = cast(Job, await self.channel.receive())
        await self._process(job)
        return job

    async def run_forever(self) -> None:
        while not self.channel.stop:
            await self.run_once()

    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
        job.apply_event(events.RUN)
        try:
            job.save()
        except Exception as save_exc:
            # Prefer persisting Failed; if that also fails, restore pre-RUN state
            # unless cancel already won (do not undo Cancelled in-memory).
            try:
                job.apply_event(events.FAIL)
                job.save()
            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}"
                )
                raise save_exc from fail_save_exc
            log.error(
                f"Job RUN persist failed: id={job.id}, error={save_exc}; "
                f"persisted Failed instead (do not re-run workflows)"
            )
            raise

        try:
            workflows = getattr(job, "workflows", None) or []
            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 
54
55
56
57
58
59
60
async def run_once(self) -> Job | None:
    if self.channel.stop:
        return None

    job = cast(Job, await self.channel.receive())
    await self._process(job)
    return job

run_forever async 

run_forever() -> None
Source code in pypepper/scheduler/worker.py 
62
63
64
async def run_forever(self) -> None:
    while not self.channel.stop:
        await self.run_once()

Executor

pypepper.scheduler.executor

Task executor interfaces and callable adapter.

IExecutor

Source code in pypepper/scheduler/executor.py 
14
15
16
17
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 
15
16
17
@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 
20
21
22
23
24
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 
23
24
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 
27
28
29
30
31
32
33
34
class CallableExecutor(Executor):
    """Executor that runs a provided callable."""

    def __init__(self, func: Callable[..., Any]):
        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 
33
34
def execute(self, task: Task, context: Context | None = None) -> Any:
    return self._func(task, context)

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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class IJobStore(metaclass=ABCMeta):
    """Pluggable persistence for JobRecord snapshots."""

    @abstractmethod
    def put(self, record: JobRecord) -> None:
        """Upsert by ``id``. Must not overwrite an existing row's ``created``."""
        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) -> None

Upsert by id. Must not overwrite an existing row's created.

Source code in pypepper/scheduler/store/interfaces.py 
26
27
28
29
@abstractmethod
def put(self, record: JobRecord) -> None:
    """Upsert by ``id``. Must not overwrite an existing row's ``created``."""
    pass

get abstractmethod 

get(job_id: str) -> JobRecord | None
Source code in pypepper/scheduler/store/interfaces.py 
31
32
33
@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 
35
36
37
@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 
39
40
41
@abstractmethod
def list(self, channel_id: str | None = None) -> list[JobRecord]:
    pass

clear abstractmethod 

clear() -> None
Source code in pypepper/scheduler/store/interfaces.py 
43
44
45
@abstractmethod
def clear(self) -> None:
    pass

JobRecord dataclass

Serializable job snapshot (metadata only; no executors).

Source code in pypepper/scheduler/store/interfaces.py 
 9
10
11
12
13
14
15
16
17
18
19
20
@dataclass(frozen=True)
class JobRecord:
    """Serializable job snapshot (metadata only; no executors)."""

    id: str
    category: str | None
    channel_id: str
    status: str
    created: str
    updated: str
    workflow_count: int = 0
    version: int = 1

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

InMemoryJobStore

Bases: IJobStore

Thread-safe dict-backed job store.

Source code in pypepper/scheduler/store/memory.py 
11
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
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) -> None:
        with self._lock:
            existing = self._store.get(record.id)
            if existing is not None:
                record = replace(record, created=existing.created)
            self._store[record.id] = record

    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) -> None
Source code in pypepper/scheduler/store/memory.py 
18
19
20
21
22
23
def put(self, record: JobRecord) -> None:
    with self._lock:
        existing = self._store.get(record.id)
        if existing is not None:
            record = replace(record, created=existing.created)
        self._store[record.id] = record

get 

get(job_id: str) -> JobRecord | None
Source code in pypepper/scheduler/store/memory.py 
25
26
27
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 
29
30
31
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 
33
34
35
36
37
38
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 
40
41
42
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)