Appendix E: Graph Manager Classes

Previous Chapter: Appendix D: Crawl Manager Classes

---

• Introduction
• Class hierarchy
• GraphManager
• Tasks: Task and SpiderTask
• Dependencies, on_finish and retries
• Parallelization
• Resources
• Writing and running a graph manager

Introduction

A graph manager runs an arbitrary workflow (a directed graph) of tasks on Scrapy Cloud — spiders, scripts (including crawl managers and deliver scripts), with dependencies between them. GraphManager (shub_workflow/graph/__init__.py) is a WorkFlowManager (see Appendix C: Workflow Manager) — so it inherits the name / flow-id / owned-jobs / resume machinery — that schedules tasks (shub_workflow/graph/task.py) according to a graph you declare in configure_workflow().

This appendix is the complete reference. For an example-led introduction read the Graph Managers tutorial chapter first.

Typical use: "run crawl manager X, then once it finishes run deliver script Y" — but the graph can be any DAG, with fan-out, fan-in, conditional next-steps by outcome, retries, per-task parallelism and resource limits.

Class hierarchy

WorkFlowManager                      (Appendix C)
      ▲
GraphManager                         orchestrates a DAG of tasks declared in configure_workflow()

BaseTask (abc)                       (shub_workflow/graph/task.py)
   ├── Task                          schedules a SCRIPT (command + args; Jinja template → parallel)
   └── SpiderTask                    schedules a SPIDER

GraphManager

You subclass it and override configure_workflow(), which builds the task graph and returns a tuple of the root tasks. Linking tasks (add_next_task, add_wait_for) defines the edges; the manager then schedules tasks as their dependencies are satisfied, until nothing is pending or running. It's a loop manager, so set loop_mode.

from shub_workflow.graph import GraphManager
from shub_workflow.graph.task import Task


class MyGraphManager(GraphManager):
    loop_mode = 120

    def configure_workflow(self):
        crawl = Task("crawl", "py:crawlmanager.py", init_args=["example.com"])
        deliver = Task("deliver", "py:deliver.py", init_args=["spider:example.com", "--output-file=s3://b/out.jl"])
        crawl.add_next_task(deliver)        # deliver runs after crawl completes
        return (crawl,)                     # root task(s)

Starting the workflow

A graph manager must be told where to start (on_start errors otherwise):

• --root-jobs — start from the tasks returned by configure_workflow().
• --starting-job <task_id> / -s <task_id> (repeatable) — start from specific task ids.

You can't pass both. Other tasks run as they're reached via the graph edges.

CLI arguments added

--root-jobs, --starting-job/-s (repeatable), --only-starting-jobs (don't follow on_finish edges), --skip-job <task_id> (repeatable; also skips its downstream), --jobs-graph (define the whole graph as YAML on the command line instead of configure_workflow), --comment (a no-op to vary the command-line signature so two instances can run concurrently). Plus everything from WorkFlowManager (positional name, --max-running-jobs, --loop-mode, --flow-id, …).

Key methods

Method	Purpose
configure_workflow(self) -> Tuple[BaseTask, ...]	abstract — build the graph; return root tasks.
workflow_loop(self)	one cycle: check running jobs, then start pending jobs whose dependencies are met.
bad_outcome_hook(self, task_id, jobid, outcome)	override — react to a task that failed and has no retries left (e.g. alert).
get_task(task_id) / pending_jobs (property) / get_running_jobid(task_id)	inspect graph state.
run_job / run_pending_jobs / check_running_jobs / handle_retry	the scheduling internals (rarely overridden).

Resuming is inherited from WorkFlowManager: each scheduled job is tagged TASK_ID=<id>, so on restart the manager rebuilds its running/completed maps for the same flow id.

Tasks: Task and SpiderTask

Tasks live in graph/task.py. All derive from BaseTask and share these constructor params: task_id (unique; can't be "retry"), tags, units, retries (default 1), project_id (run in another project), wait_time (don't start until N seconds after going pending), on_finish.

Task class	Schedules	Key params	Notes
Task(task_id, command, init_args=, retry_args=, …)	a script via schedule_script	command = script name (py:foo.py) or a Jinja2 template; init_args appended; retry_args used instead on retry	a multi-line rendered command ⇒ parallel subtasks (see below). . is not allowed in task_id.
SpiderTask(task_id, spider, …, job_settings=, **spider_args)	a spider via schedule_spider	spider name; **spider_args; optional job_settings	no parallelization.

Linking (graph edges)

Call these on a task before the manager locks the graph (i.e. inside configure_workflow):

Method	Meaning
add_next_task(task)	task becomes a successor — runs after this one completes (the "default" on_finish edge).
add_wait_for(task)	this task must wait for task to complete (a pure dependency edge, no scheduling-after).
add_required_resources(ResourcesDict({Resource("x"): n}))	this task needs n of resource x to start (see Resources).
set_start_callback(fn)	advanced — called when the task goes pending; lets you add successors lazily.

Dependencies, on_finish and retries

A pending task starts only when all its wait_for dependencies have completed, its wait_time has elapsed, required resources are available, and max_running_jobs isn't reached. The manager finishes when there are no pending and no running jobs.

on_finish routing. Each task has an on_finish dict mapping an outcome to a list of next task ids (or the literal "retry"). When a task finishes, the next jobs are chosen as:

1. an entry for the exact close-reason outcome, else
2. the "failed" entry if the outcome is in failed_outcomes (inherited from WorkFlowManager), else
3. the "default" entry — which is exactly the add_next_task successors.

add_next_task populates "default"; and if retries > 0, "failed" defaults to ["retry"]. So out of the box a failed task is retried up to retries times, then (no retries left) routed through bad_outcome_hook. Provide a custom on_finish (constructor arg) to branch on specific outcomes — e.g. on_finish={"finished": ["deliver"], "no_items": ["alert"]}.

Retries are per task (retries, default 1). On a failed outcome with retries remaining the task is re-queued (decrementing its counter); retry_args (on a Task) replace init_args for the retry.

Parallelization

A Task whose command template renders multiple lines is split into that many parallel subtasks (<task_id>.0 … <task_id>.N-1), each scheduled independently; successors wait for all subtasks. This is the idiomatic way to fan a task out:

Task("crawl", "{% for i in range(4) %}py:crawlmanager.py --part={{ i }}\n{% endfor %}")
# → 4 parallel jobs: crawl.0 … crawl.3

SpiderTask does not parallelize (get_parallel_jobs() is always 1).

Resources

Resources are named semaphores that gate how many tasks of some class run at once, independently of max_running_jobs. Declare a task's needs with add_required_resources(ResourcesDict({Resource("db"): 1})); the manager auto-sizes the available amount to the largest single requirement, and a task only starts once it can acquire all its resources (released when it finishes). Use it to, say, allow only one "heavy" task at a time while many light ones run freely. For parallelized tasks the requirement is split across subtasks (fractions), so the whole task uses the declared amount in aggregate.

Writing and running a graph manager

import time
import logging

from shub_workflow.graph import GraphManager
from shub_workflow.graph.task import Task, SpiderTask


class MyGraphManager(GraphManager):

    loop_mode = 120          # required (it's a loop manager)
    # name comes from the positional arg or a `name` attribute (WorkFlowManager)

    def configure_workflow(self):
        discover = SpiderTask("discover", "example_spider", category="news")
        crawl = Task("crawl", "py:crawlmanager.py", init_args=["example.com"], retries=2)
        deliver = Task("deliver", "py:deliver.py",
                       init_args=["spider:example.com", f"--output-file=s3://b/{int(time.time())}.jl"])

        discover.add_next_task(crawl)
        crawl.add_next_task(deliver)
        return (discover,)         # root task(s)


if __name__ == "__main__":
    from shub_workflow.utils import get_kumo_loglevel

    logging.basicConfig(format="%(asctime)s %(name)s [%(levelname)s]: %(message)s", level=get_kumo_loglevel())
    MyGraphManager().run()

Invoke it with a name and a start directive:

python flowmanager.py mygraph --root-jobs

A graph manager is a normal shub-workflow script: register the file in your project's setup.py so it deploys and runs on Scrapy Cloud as py:flowmanager.py. The scripts/spiders it schedules are separate deployables. A common companion is a delivery issuer (IssuerScriptWithSCJobInput, targeted at the crawl's spider as spider:<name>) that merges and delivers the items of one workflow instance — filtering the read jobs on the manager's FLOW_ID tag so it picks up all and only this workflow's jobs (see the Graph Managers tutorial for a worked example). It replaces the deprecated BaseDeliverScript.

The helper get_scheduled_jobs_specs(manager, job_ids) (graph/utils.py) reads a task job's logs to recover the (kind, name, job id) of the jobs it scheduled — useful when a task is itself a manager.

---

Next Chapter: Appendix F: Monitor Classes