Appendix G: Issuer Classes

Previous Chapter: Appendix F: Monitor Classes

---

• Introduction
• Class hierarchy
• The item model
• IssuerScript
• Output: slots, batches, filenames
• Deduplication
• The two ready-made input subclasses
• IssuerScriptWithSCJobInput options
• Consumers and deduplicators
• Configuration attributes
• Delivery as an issuer
• Writing and running an issuer

Introduction

An issuer is a fundamental Scrapy Cloud workflow component: it reads a massive input (finished spider jobs, or batch files in a folder), processes the items (dedup, filter, transform, split, re-tag…), and writes a massive output as batch files. Chaining issuers builds the whole post-crawl data-processing pipeline — consumers, deduplicators, balancers, and final delivery are all issuers. IssuerScript (shub_workflow/issuer/__init__.py) is the base class.

It is a BaseLoopScript (see Appendix B: Script Classes): it loops, reading new inputs and flushing output batches, until its inputs are exhausted (or forever, if it's a continuous stage). It is generic over two type parameters: the item type (ITEMTYPE, a subtype of IssuerItem) and the per-input args passed from input discovery to input processing (PROCESS_INPUT_ARGS_TYPE).

Issuers also supersede the old delivery base class — see Delivery as an issuer.

Class hierarchy

BaseLoopScript                          (Appendix B)
      ▲
IssuerScript[ITEMTYPE, PROCESS_INPUT_ARGS_TYPE]     read input → process/dedup → write batch output
      ▲
      ├── IssuerScriptWithFileSystemInput[ITEMTYPE]        input = batch files in a folder
      └── IssuerScriptWithSCJobInput[ITEMTYPE]             input = finished Scrapy Cloud spider jobs

The item model

Items flowing through an issuer are IssuerItem TypedDicts:

Field	Meaning
id	unique id, computed by your build_item_id() and used for dedup.
source	the source crawler / canonical name the item belongs to.
input_source	where the item was read from (a job key or input file name).
search_keywords	(optional) set of keywords the item came from; merged when the same id recurs.

Subclass IssuerItem to add your own fields and bind the issuer to it (class MyIssuer(IssuerScript[MyItem, Tuple[()]])). You must implement build_item_id(item) -> ItemId. Override adapt_input_item(raw) -> ITEMTYPE if the raw input records aren't already in your item shape (default just casts).

IssuerScript

The loop & data flow

workflow_loop() (provided) calls get_new_inputs() → for each input, process_input(), which iterates the input's records and calls process_item() per record (or, when explode_input_items is set, once per object the jmespath selects inside each record). process_item():

1. stamps input_source, computes id = build_item_id(item), counts totals;
2. if dedup is on and the id was already seen, drops it (counts a dupe);
3. otherwise issue_item() — enqueue into items_queue[slot][source] (in memory by default, or on disk with persist_items_queue_on_disk), and when that queue reaches the item's filesize, send_file() flushes it to a gzipped JSON-lines batch file.

On close it flushes remaining queues and removes/marks the fully-processed inputs. An input is only removed once all the items it contributed have been written out (tracked in pending_inputs_to_remove).

What you implement

Method	Required?	Purpose
build_item_id(item) -> ItemId	abstract	the dedup/identity key for an item.
get_new_inputs() -> Iterable[(InputSource, args)]	abstract*	discover inputs to process (the ready subclasses implement this).
process_input(inputsrc, args) -> bool	abstract*	read one input, call process_item() per record; return True if processed.
remove_inputs(inputs)	abstract*	consume/retire processed inputs (the subclasses implement this).
adapt_input_item(raw)	optional	adapt raw records to your ITEMTYPE.
get_output_slot_for_item / get_filesize_from_item / compute_destination_filename	optional	customize output routing / batch size / naming.

* implemented for you by the two input subclasses below; you usually only write build_item_id (and a custom IssuerItem) plus set a few attributes.

Output: slots, batches, filenames

Items are queued per (output slot, source) and flushed to a file when the queue reaches the item's filesize (default_filesize, or per-item via get_filesize_from_item). Output goes to output_folder; filenames are timestamped and prefixed by source (and slot / input_slot when set), e.g. <output_slot>_<source>_<YYYYmmddThhmmss.ffffff>_<input_slot>.jl.gz.

Output slots enable parallel downstream processing: set parallel_outputs = N and each item is routed to slot hash_mod(id, N) (so the same id always lands in the same slot — letting N parallel downstream issuers each own a slot). Or pin a single output_slot. input_slot similarly selects one input slot for this instance (N parallel instances, one per slot).

Deduplication

When dedupe is on (default), process_item() skips ids already in a bloom filter (seen, capacity MAX_ITEMS, error rate ERRORS_RATE, persisted to a livedup.bloom file). Plain in-memory dedup only covers the job's own lifetime — fine for a short-lived stage, lost on restart.

For long-term dedup that survives restarts, set LOAD_DELIVERED_IDS_DAYS and call load_last_outputs(output_folders, …) in __init__ — on start it reads the output files (selected by their timestamped names) from the last N days back into seen. output_folders is a tuple, so it can include downstream folders (where items land after further processing), not just this issuer's own output — ensuring an id already passed along stays deduplicated. This is what makes a deduplicator's dedup persistent across restarts (see Consumers and deduplicators). If you set LOAD_DELIVERED_IDS_DAYS but never call load_last_outputs, on_start() raises (a guard against silently-incomplete dedup). Bloom filters have a small false-positive rate (a unique item dropped as a dupe) — acceptable for throughput-oriented pipelines.

The two ready-made input subclasses

Pick one; it implements get_new_inputs / process_input / remove_inputs so you only write build_item_id (+ output attrs):

• IssuerScriptWithFileSystemInput — input is batch files in input_folder (optionally filtered by input_slot prefix). Each file is a gzipped JSON-lines list; processed inputs are moved to processed_folder if set, else removed. The typical mid-pipeline stage (consumer → filter → deduplicator → balancer).
• IssuerScriptWithSCJobInput — input is finished Scrapy Cloud spider jobs. A required positional CLI arg target (spider:<name> | canonical:<name> | class:<ClassName>) selects which spiders' jobs to read; their items are processed and the consumed jobs are tagged CONSUMED=True (so they aren't re-read). The typical first stage (reading raw crawl output). The matched spiders are visited round-robin, rotating the starting spider across loops (only max_inputs_per_loop inputs run per loop, and each loop resumes where the previous one stopped), so when the target matches many spiders — e.g. class:<BaseSpider> to consume every source in one job — a source with a large backlog cannot starve the others; it is a no-op for a single-spider target. It also exposes a couple of extra flags — see the next section.

IssuerScriptWithSCJobInput options

Beyond the generic issuer attributes, IssuerScriptWithSCJobInput adds two flags:

Attribute	Default	Meaning
set_item_source	True	Stamp each read item's source with the scanned spider's canonical name. Set to False when the scanned spider is a secondary / post-processing stage — one that processes data already produced by a primary spider. Such items already carry their originating source, which must be conserved rather than overwritten. (This is exactly why a delivery issuer, which reads a secondary spider, keeps the upstream source.)
flush_on_each_input	False	When True, flush_files() runs at the end of processing each scanned job (instead of only when a source's spider stops or on close), and the just-flushed job is consumed (tagged CONSUMED) immediately rather than at the end-of-loop sweep — so if the script is killed mid-loop, jobs already written out are not re-delivered on restart. Consuming an input also uploads stats at that moment (see the note below the table), so a monitor still sees the delivery stats of every delivered job even when the script is later killed. Some pipelines instead deliver several jobs into one output batch and leave this False.
scope_input_to_flow_id	False	When True, read only input jobs tagged with this script's own FLOW_ID (from --flow-id, or the FLOW_ID tag when running inside a workflow), so a script scheduled by a graph manager reads all and only the jobs of its own workflow instance. Ports the workflow-scoping the deprecated BaseDeliverScript did automatically. No-op (reads everything, with a warning) if no flow_id is set.

One output file per scanned job. flush_on_each_input guarantees a flush per job, but a batch can still be split earlier if it reaches default_filesize. To ensure all items of a single job land in the same output file, pair flush_on_each_input=True with a large default_filesize (or a get_filesize_from_item() returning a big number for the relevant items).

process_input() also calls a post_process_input_items(spider_job, args) hook (default a no-op) once per scanned job, after all of that job's items have been read (and processed via process_item()) and before the optional per-job flush. spider_job is the just-read Scrapy Cloud job — use its metadata / key / items (the key is also args[0]["key"]). Override it to run any per-job finalization logic.

Its most common use is the accumulate-then-merge pattern, which is generic to any IssuerScriptWithSCJobInput (delivery or not): when an input's records must be combined into fewer output records (a join, roll-up, or reconciliation), override process_item() to accumulate the records instead of issuing them inline; then in post_process_input_items() merge them, set each combined record's id / input_source, issue_item() the results, and reset the accumulator. Paired with flush_on_each_input=True, the merged records are written as a single output file per job. A delivery issuer may use this pattern, but so may a filter or a roll-up stage — it is not a property of delivery.

The hook is also the place to aggregate a scanned job's stats if you need it: there is no built-in flag for that — mix in SpiderStatsAggregatorMixin yourself and call self.aggregate_spider_stats(...) from your post_process_input_items() override.

Consumers and deduplicators

Two issuer archetypes recur, making opposite trade-offs around dedup persistence:

• Consumer — the first stage. Reads spider jobs, does a cheap, in-memory preliminary dedup, and distributes items across parallel_outputs slots (the same id always lands in the same slot, so dedup can be parallelized downstream), often also extracting seeds to the frontier. A consumer is short-lived and cheaply restartable: it does not use load_last_outputs, so a restart loses only its in-memory dedup — acceptable, because the authoritative dedup is downstream. It's the right place for post-processing logic that changes over time.
• Deduplicator — the heavy stage. Runs continuously, typically one instance per slot, doing the massive, long-term deduplication. It survives restarts: on start it calls load_last_outputs(...) to refill its bloom filter from the last LOAD_DELIVERED_IDS_DAYS of its own (and downstream) output, so previously-emitted ids stay deduplicated. Deduplicators usually have a per-instance capacity (MAX_ITEMS) and run until it's reached.

Why not do the massive dedup in the consumer? Two reasons. (1) The consumer is a single process that also writes seeds and holds the post-processing logic that changes over time, so it must be restartable cheaply — doing the persistent massive dedup there would make every restart reload an enormous id history (and risk the dedup). (2) Massive dedup must scale beyond one process: the consumer instead fans items out by id-hash across parallel_outputs slots, so N deduplicators each dedup one slot independently (scale further by adding slots). Keeping them separate lets the volatile consumer be stopped/updated/rescheduled without affecting the authoritative, persistent deduplication.

Configuration attributes

Attribute	Default	Meaning
output_folder	(required)	where batch output files are written.
default_filesize	10_000	items per output batch file.
parallel_outputs	1	number of output slots (>1 ⇒ hash-routed slots).
separate_output_by_source	True	one output file per (slot, source). False ⇒ all sources of a slot share one queue and pack into default_filesize-sized files (one file per slot), the source is dropped from the default filename, and the per-source stopped-spider partial flush is disabled (a slot's batch flushes only at default_filesize or on close).
persist_items_queue_on_disk	False	each output-queue bucket (items_queue[slot][source]) is an in-memory dict by default — fast, but a whole batch is held in RAM until flushed. True ⇒ back each bucket with an on-disk SqliteDict: items live on disk and are read back in bounded chunks when the output file is written, so memory stays bounded (slower). Use it when items are large enough that a whole batch would not fit in RAM — e.g. a metadata-heavy delivery.
persist_items_queue_dir	None (cwd)	directory holding the persist_items_queue_on_disk sqlite files. Defaults to the current working directory rather than tempfile's default /tmp, so a big batch does not depend on how much room /tmp happens to have; point it at another filesystem when you need more space.
items_queue_read_chunk_size	100	how many items are read at a time from an on-disk bucket while writing the output file. Peak memory during a flush is roughly this many items, so lower it for very large items (100 items of 100 KB ≈ 10 MB). No effect unless persist_items_queue_on_disk is set — see the memory note below the table.
explode_input_items	None	a jmespath expression selecting a list inside each raw input record; when set, process_item() is called once per selected object (one record → many items) instead of once per record. Lets a spider whose records bundle several issuable objects be consumed without overriding process_input().
input_slot / output_slot	None	pin this instance to one input / output slot.
dedupe	True	enable bloom-filter de-duplication.
MAX_ITEMS	200_000_000	bloom capacity; also the loop's processed-items ceiling.
ERRORS_RATE	1e-8	bloom false-positive rate.
LOAD_DELIVERED_IDS_DAYS	(unset)	days of prior output to reload into seen (needs load_last_outputs).
close_on_no_inputs	False	stop when there are no more inputs (vs. keep looping).
max_inputs_per_loop	-1	cap inputs processed per loop cycle (-1 = unlimited).
min_wait_time_secs_to_flush_stopped_spiders	0	grace before flushing pending items of a source whose spider has stopped.
loop_mode	0	(inherited) seconds between cycles; set it for a continuous issuer.

Coupling note: each loop, if a source still has queued items, the issuer checks get_project_running_spiders(crawlmanagers=("py:crawlmanager.py",)); if that source's spider is no longer running it flushes the partial batch after min_wait_time_secs_to_flush_stopped_spiders. So the default expects the upstream discovery crawl manager to be named py:crawlmanager.py.

Stats/consumption coupling: whenever inputs are consumed (an input is retired once all its items have been written), the issuer uploads stats at that same moment, in addition to the inherited periodic and on-close uploads. This keeps a monitor's stats consistent with what was actually delivered: a consumed input is never reprocessed, so if the script is killed before its next periodic/close upload, the delivery stats of those inputs would otherwise be lost — leaving the monitor with no stats for jobs that were in fact delivered.

Memory / OOM: the output queue buffers a whole (slot, source) batch until it reaches default_filesize (or, with flush_on_each_input, the end of the input). With large items and large batches — e.g. a metadata-heavy delivery using flush_on_each_input + a huge default_filesize — that batch can exceed the container's memory and the job dies with close_reason: "killed by oom". The remedies are to make each item smaller, lower default_filesize, or set persist_items_queue_on_disk=True so the batch is held on disk instead of in RAM. Note the batch then needs roughly its own size in disk space, for the sqlite file plus the compressed output file staged locally before it is moved to its destination.

Never iterate an on-disk bucket with sqlitedict's values() / items() / keys(). They look lazy but are not: each hands a single whole-table SELECT to sqlitedict's writer thread, which pushes every row into an unbounded in-memory queue as fast as sqlite yields them, with no back-pressure from the consumer — its own select() docstring says "the entire result will be in memory". Iterating a disk-backed batch that way holds all of it in RAM, which defeats persist_items_queue_on_disk completely and OOM-kills the job even though the items really are on disk. Read in bounded chunks instead: _iter_bucket_values() in shub_workflow/issuer/__init__.py pages through a bucket keyset-paginated by rowid, holding at most items_queue_read_chunk_size items at a time — use it for any new bucket iteration. Relatedly, count with len(bucket) (a COUNT(*)) and never len(bucket.keys()): on an on-disk bucket keys() is a generator, so that raises a TypeError, and it would read the whole table just to count it.

Delivery as an issuer

The final delivery stage is now built as an issuer. This replaces the older shub_workflow.deliver.BaseDeliverScript, which is deprecated — instantiating it now emits a DeprecationWarning. New delivery scripts, and migrations of old ones, should be issuers.

There is no dedicated delivery class. A delivery is simply an IssuerScriptWithSCJobInput configured for the terminal stage — the base class already provides everything a delivery needs. Which attributes you set is use-case dependent, not fixed by "delivery":

• close_on_no_inputs = True — typical: deliver the finished jobs, then stop rather than loop forever. Override for a continuously-running delivery.
• flush_on_each_input = True + a large default_filesize — when you want one delivery file per scanned job (the common case). Leave flush_on_each_input at False if a delivery batches several jobs into one output file.
• dedupe = False — set this only when the delivered data was already deduplicated upstream (a post-processing pipeline: consumer → deduplicator → … → delivery). A delivery that reads a primary spider's crawl directly may keep dedupe = True (or reproduce the old field-based dedup).
• set_item_source = False — set this only when the delivered spider is a secondary / post-processing stage whose items already carry their originating source. Delivering a primary spider keeps the default (True), so source is stamped with that spider's canonical name.

You always supply:

• build_item_id() — required (abstract); issue_item() uses the id for output queueing/routing even when dedupe = False.
• the delivery destination — via output_folder and/or a compute_destination_filename(output_slot, source) override. With flush_on_each_input = True there is a natural per-job → per-file mapping; the destination is often derived from the scanned job's spider args (read them in post_process_input_items()).

Delivery issues each read item as-is by default. Combining records before writing is not a property of delivery: if a particular delivery needs it, it uses the generic accumulate-then-merge pattern via post_process_input_items(), exactly as any other issuer would.

Migrating from BaseDeliverScript

Rebuild the delivery as an IssuerScriptWithSCJobInput (no dedicated class) and choose the configuration above to fit your pipeline. The old hooks map over as follows:

• on_item(item, scrapername) → process_item(item, input_source) (issue inline, or accumulate);
• process_job_items(scrapername, job) → the base process_input() reads the job, and per-job finalization goes in post_process_input_items(spider_job, args);
• uploading a per-job file via the filesystem helper → issue_item() + a per-job flush (flush_on_each_input = True), with the path from compute_destination_filename();
• DEDUPE_KEY_BY_FIELDS / is_seen_item() → dedupe + build_item_id() (bloom filter);
• the scrapername positional arg → the target arg (spider: / canonical: / class:);
• DELIVERED_TAG = "delivered" (which marked done jobs) → the issuer selects jobs lacks_tag=CONSUMED_TAG and tags consumed ones with CONSUMED_TAG (default "CONSUMED=True").

Watch out for: preserving the "delivered" tag (production-critical) — BaseDeliverScript tagged delivered jobs "delivered", but the issuer defaults to CONSUMED_TAG = "CONSUMED=True"; deploying the migration with the default would re-deliver every already-delivered job (none carry CONSUMED=True), so set CONSUMED_TAG = "delivered" on the migrated issuer; the invocation changes (deliver.py <scrapername…> → deliver.py <type>:<name>, so update the scheduling job); FLOW_ID scoping — the issuer reads all finished, un-consumed jobs of the target, so if a graph manager used to scope delivery to its workflow instance, set scope_input_to_flow_id = True (which the deprecated class did automatically); conserving the source for a secondary spider (set_item_source = False, or the base overwrites item["source"]); reading the per-job destination from the job's spider args in post_process_input_items(); and converting scrapy.Items to plain dicts before issuing. For per-job stats aggregation there is no built-in flag — mix in SpiderStatsAggregatorMixin and call aggregate_spider_stats(...) in the hook.

💡 Claude assistance. shub-workflow ships a Claude Code plugin, shub-workflow-toolkit, whose shub-workflow-issuers skill teaches Claude how to build issuers and carry out this BaseDeliverScript → issuer migration (it bundles a dedicated migration reference covering the configuration choices, method mapping and gotchas above). Install it as described in the repository README and Claude will be able to assist you with the migration.

Writing and running an issuer

A minimal file-input issuer (a mid-pipeline consumer/deduplicator):

import logging
from typing import Tuple
from shub_workflow.issuer import IssuerScript, IssuerScriptWithFileSystemInput, IssuerItem, ItemId


class MyItem(IssuerItem):
    url: str


class MyIssuer(IssuerScriptWithFileSystemInput[MyItem]):

    loop_mode = 60
    input_folder = "gs://my-bucket/stage-in"
    output_folder = "gs://my-bucket/stage-out"
    parallel_outputs = 10                 # 10 hash-routed output slots
    LOAD_DELIVERED_IDS_DAYS = 30          # dedup against the last 30 days of output

    def __init__(self):
        super().__init__()
        self.load_last_outputs((self.output_folder,))   # required when LOAD_DELIVERED_IDS_DAYS is set

    def build_item_id(self, item: MyItem) -> ItemId:
        return ItemId(item["url"])


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())
    MyIssuer().run()

As with any shub-workflow script, register the file in your project's setup.py so it deploys and runs on Scrapy Cloud as py:myissuer.py; deployment is out of scope here.