overview-plugin-workqueue 1.0.1

In-process work queue for use in Overview plugins

overview-plugin-workqueue

In-process work queue for use in Overview plugins.

The problem it solves

You’re writing an Overview plugin. It’s a web server. And when a user creates a View of your plugin, you want to kick off some long-running computation.

overview-plugin-workqueue will solve this problem in a “prototype” or lightweight situation. It’s meant to be used on a single Python server.

Your plugin’s architecture

Your plugin should be a web server with four endpoints:

• GET /metadata: the usual metadata

• GET /show: present HTML with JavaScript web app. The JavaScript will call:

• GET /search?server=...&documentSetId=... (or some-such): read Authorization header with api_token; generates some JSON using a locally-stored “database”.

But this “database” – specific to the apiToken on this server – must exist! So we need:

• POST /generate: accept form data with server and document_set_id parameters, plus an Authorization header (or api_token parameter); generates the database for this view.

The Python side of things

Your plugin is a web server. Remember: Web servers are stateless.

Your plugin should save and read a “database” file whose name is derived from server, document_set_id and api_token. Don’t worry – workqueue will name the file for you.

You must anticipate spurious /generate calls, and you must give progress reports to your users or they’ll assume something’s wrong. You also want jobs to continue even if your users go away; and when they reconnect, progress should resume where it left off.

This is where workqueue shines. It uses an in-process thread pool to limit the number of concurrent calls, and it uses generators to keep the server responsive even when serving long-polling progress-report requests.

Install it: pip3 install workqueue

Now create a database-building script as a separate executable. For instance, process-docset.py (chmod +x it, and make sure it begins with #!/usr/bin/env python3.) workqueue will invoke it like this:

./process-docset.py SERVER DOCUMENT_SET_ID API_TOKEN OUTPUT_FILENAME

And workqueue will provide these arguments:

• SERVER: Overview server – e.g., https://www.overviewdocs.com.

• DOCUMENT_SET_ID: String ID – e.g., 1234.

• API_TOKEN: An alphanumeric API token.

• OUTPUT_FILENAME: An empty tempfile the script must overwrite.

Test your script by running it on the command line.

Now you’re ready to use workqueue in your plugin web server. For example, here it is in Flask:

from pathlib import Path
from typing import Mapping

from werkzeug.wrappers import Response
from workqueue import WorkQueue, OverviewViewParams

# ... set `app` ... and then

WORK_QUEUE = WorkQueue(
        Path(__file__).parent / "process-docset.py",
        executor=ThreadPoolExecutor(2, "process-docset-"),
        storage_dir=Path(__file__).parent / "databases",
)

def _extract_api_token(auth_header: str) -> str:
    m = _AUTH_HEADER_REGEX.match(auth_header)
    if not m:
        raise ValueError(
            'Authorization header must look like "Basic [base64-encoded data]"'
        )
    decoded = base64.b64decode(m.group(1))
    m = _AUTH_TOKEN_STRING_REGEX.match(decoded)
    if not m:
        raise ValueError(
            'base64-encoded Basic HTTP Auth value must look like "[api_token]:x-api-token"'
        )
    return m.group(1).decode("latin1")


def _extract_params(
    form: Mapping[str, str], headers: Mapping[str, str]
) -> OverviewViewParams:
    try:
        return OverviewViewParams(
            form["server"],
            form["documentSetId"],
            _extract_api_token(headers["Authorization"]),
        )
    except (KeyError, ValueError) as err:
        raise BadRequest(
            "Expected form data ?server=...&documentSetId=... and Authorization header. "
            "Problem: %s" % str(err)
        )


@app.route("/generate", methods=["POST"])
def generate():
    """
    Stream a JSON Array of "progress" events, as a long-polled response.

    When the JSON stream is finished, the database is ready.

    This method returns early, passing Flask a generator. Flask will consume
    the chunks from the generator and send them to the client.
    """
    overview_params = _extract_params(request.form, request.headers)

    job = WORK_QUEUE.ensure_run(overview_params)
    if job is None:
        # The work is done; the model can be loaded now.
        progress_stream = []
    else:
        progress_stream = WORK_QUEUE.report_job_progress_until_completed(job)

    # Stream progress to the user, as a JSON Array. (Must not be buffered.)
    json_stream = itertools.chain(
        ["["],
        (
            # JSON array: comma between elements, not before first or after last
            ("" if i == 0 else ",") + json.dumps(progress._asdict())
            for i, progress in enumerate(progress_stream)
        ),
        ["]"],
    )

    return Response(json_stream, content_type="application/json")

The JavaScript side of things

On app startup, POST to /generate using Oboe. Pass it server and documentSetId, and pass an Authorization header (or apiToken form parameter) as well.

Use Oboe, not a normal XMLHttpRequest or Fetch request, because Oboe will notify about progress events before /generate completes. (And /generate could take a very long time indeed to complete.)

Once /generate completes, you can invoke /query using the same server, documentSetId and apiToken.

Scaling limits

workqueue is designed to be used in relatively-small deployments. It’s a great place to start. Here are the parts that will limit you:

• storage: workqueue is designed to read and write the local disk. TODO read and write S3-compatible storage servers.

• single-server: workqueue cannot distribute load across multiple machines. Use a different process to achieve that.

Developing This Package

Install dependency: pip3 install --user tox

Run tox to make sure it works for you.

Then the development loop:

1. Write a test. Run tox and prove to yourself it fails.

2. Write code. Run tox to prove it solves your problem.

3. Submit a pull request.

Release Process

1. Edit the version= line in setup.py. Adhere to semver.

2. git commit setup.py -m "vX.X.X" && git push

3. python ./setup.py sdist && twine check dist/* && twine upload dist/*-X.X.X.tar.gz

License

MIT. See LICENSE.