API reference

Utilities

process.util.wrap(string)

Format a long string as a help message, and return it.

process.util.walk(paths)

process.util.get_publisher()

process.util.consume(*args, **kwargs)

process.util.decorator(decode, callback, state, channel, method, properties, body)

Close the database connections opened by the callback, before returning.

If the callback raises an exception, shut down the client in the main thread, without acknowledgment. For some exceptions, assume that the same message was delivered twice, log an error, and ack the message. Nack without requeuing if the message requires review (e.g. unexpected messages), for any future dead-letter exchange.

process.util.get_or_create(model, data)

Get or create a PackageData or Data instance.

process.util.lock_collection(collection_id)

Lock the collection row with KEY SHARE, and return the collection, or None if it is deleted.

Call this within a transaction that writes rows referencing the collection. KEY SHARE conflicts only with the wiper worker’s FOR UPDATE; as such, other concurrent workers don’t block each other.

A worker can call this only if a single transaction encloses all its writes, and if it can abandon that transaction on a None return value. Workers that can’t call this:

• compiler creates each step in autocommit mode, interleaved with publishing messages.

• file_worker can’t abandon its transaction: deleting_step() would create a CHECK step for the collection.

process.util.create_note(collection, code, note, **kwargs)

process.util.create_step(name, collection_id, **kwargs)

process.util.deleting_step(*args, **kwargs)

Delete the named step and run any finish callback only if successful or if the error is expected.

process.util.delete_step(name, finish=None, finish_args=(), exception=None, **kwargs)

process.util.create_logger_note(collection, name)

process.util.get_extensions(package)

Models

class process.models.Default

class process.models.Collection(*args, **kwargs)

A collection of data from a source.

There should be at most one collection of a given source (source_id) at a given time (data_version) of a given scope (sample or not). A unique constraint therefore covers these fields.

A collection can be a sample of a source. For example, an analyst can load a sample of a bulk download, run manual queries to check whether it serves their needs, and then load the full file. To avoid the overhead of deleting the sample, we instead make sample part of the unique constraint, along with source_id and data_version.

class Transform(*values)

clean_fields(exclude=None)

Clean all fields and raise a ValidationError containing a dict of all validation errors if any occur.

get_upgraded_collection()

Return the upgraded collection or None.

Return type:

Self | None

get_compiled_collection()

Return the compiled collection or None, traversing the upgraded collection if needed.

Return type:

Self | None

get_root_parent()

Return the “root” ancestor of the collection.

Return type:

Self

class process.models.CollectionNote(*args, **kwargs)

A note an analyst made about the collection.

class Level(*values)

class process.models.CollectionFile(*args, **kwargs)

A file within the collection.

class process.models.ProcessingStep(*args, **kwargs)

A step in the lifecycle of collection file.

class Name(*values)

class process.models.Data(*args, **kwargs)

The contents of a release, record or compiled release.

class process.models.PackageData(*args, **kwargs)

The contents of a package, excluding the releases or records.

class process.models.Release(*args, **kwargs)

A release.

class process.models.Record(*args, **kwargs)

A record.

class process.models.CompiledRelease(*args, **kwargs)

A compiled release.

class process.models.ReleaseCheck(*args, **kwargs)

The result of checking a release.

class process.models.RecordCheck(*args, **kwargs)

The result of checking a record.

Loader

process.processors.loader.file_or_directory(path)

Check whether the path exists. Raise an exception if not.

process.processors.loader.create_collection_file(collection, filename=None, url=None)

Create file for a collection and steps for this file.

Parameters:

• collection (Collection) – collection

• filename (str) – path to file data

Returns:

created collection file

Raises:

InvalidFormError – if there is a validation error

Return type:

CollectionFile

process.processors.loader.create_collections(source_id, data_version, *, sample=False, upgrade=False, compile=False, check=False, scrapyd_job='', note='', force=False)

Create the root collection, derived collections and notes.

Parameters:

• source_id (str) – collection source

• data_version (str) – data version in ISO format

• sample (boolean) – is this sample only

• upgrade (boolean) – whether to plan collection upgrade

• compile (boolean) – whether to plan collection compile

• check (boolean) – whether to plan schema-based checks

• scrapyd_job (str) – Scrapyd job ID

• note (str) – text description

• force (boolean) – skip validation of the source_id against the Scrapyd project

Returns:

the root collection, upgraded collection and compiled_collection

Return type:

tuple[Collection, Collection, Collection]

Scrapyd

process.scrapyd.configured()

Return whether the connection to Scrapyd is configured.

Return type:

bool

process.scrapyd.spiders()

Return the names of the spiders in the Scrapyd project.

Return type:

list[str]

Command-line interface

class process.cli.CollectionCommand(stdout=None, stderr=None, no_color=False, force_color=False)

add_arguments(parser)

Add default arguments to the command.

add_collection_arguments(parser)

Add arguments specific to this command.

handle(*args, **options)

Get the collection.

handle_collection(collection, *args, **options)

Run the command.