rotifer.db.parallel.SimpleParallelProcessCursor#

class rotifer.db.parallel.SimpleParallelProcessCursor(progress=True, tries=3, batch_size=None, threads=10, *args, **kwargs)[source]#

Bases: BaseCursor

Split input into batches and fetch data using parallel processes.

Subclasses implement fetcher(), which opens a data stream for a batch of accessions, and parser(), which turns that stream into result objects. Everything else (batching, retries, bookkeeping of missing entries and the process pool) is handled here.

Parameters:

progressbool, default True: Whether to print a progress bar.
triesint, default 3: Number of attempts to download each batch.
batch_sizeint, optional: Number of accessions per batch. When not set, the input is divided evenly among the worker processes.
threadsint, default 10: Number of simultaneous worker processes.

__getitem__(accession, *args, **kwargs)[source]#

Fetch data for one batch of entries, dictionary style.

The download is attempted up to tries times. Errors that match one of the giveup patterns abort the retries and mark the affected accessions as unrecoverable.

Parameters:

accessionstr or iterable of str: Database entry identifiers.

Returns:

object: The objects created by the parser() method, or None when every attempt failed.

fetchall(accessions, *args, **kwargs)[source]#

Fetch data for all entries at once.

Input order is not preserved.

Parameters:

accessionslist of str: Database entry identifiers.

Returns:

list: Every object yielded by fetchone().

fetcher(accession, *args, **kwargs)[source]#

Open remote data as a stream.

This method is called by __getitem__ and must be implemented by subclasses.

Parameters:

accessionstr or iterable of str: Database entry identifiers.

Raises:

NotImplementedError: Always, unless overridden by a subclass.

fetchone(accessions, *args, **kwargs)[source]#

Iterate over entries as parallel downloads complete.

Results are yielded in completion order, not input order.

Parameters:

accessionslist of str: Database accessions.

Yields:

object: One result object per completed download.

parser(stream, accession, *args, **kwargs)[source]#

Build result objects from the stream generated by fetcher().

This method is called by __getitem__ and must be implemented by subclasses.

Parameters:

streamfile-like: Object returned by fetcher().
accessionstr or iterable of str: Database entry identifiers.

Raises:

NotImplementedError: Always, unless overridden by a subclass.

splitter(accessions, *args, **kwargs)[source]#

Divide the input identifiers into batches.

This method is called by fetchone() to define the datasets processed by worker() in each parallel batch.

Parameters:

accessionslist of str: Database identifiers.

Returns:

generator: Lists of at most batch_size accessions. When batch_size is not set, the input is divided evenly among the worker processes.

worker(accessions, *args, **kwargs)[source]#

Process one batch of accessions in a worker process.

Accessions are sorted and, when maxgetitem allows more than one accession per call, grouped into chunks before being handed to __getitem__.

Parameters:

accessionslist of str: Database identifiers of one batch.

Returns:

dict: A dictionary with two keys: result, the list of objects generated by __getitem__, and missing, the registry of entries that could not be retrieved.