1
\$\begingroup\$

For simplicity, I have omitted the SubProcedure class which has an execute function, and returns Progress when done, for the real code I simply iterate over these sub procedures and insert the progress into the queue instead of iterating over the progress list.

This code is in logic.py:

import random
from collections import namedtuple
import asyncio


Progress = namedtuple('Progress', ('stage', 'progress'))

class JobManager:
    _selected_job = None
    _latest_stage = None

    def __init__(self, loop):
        self._message_queue = asyncio.Queue(loop=loop)

    def current_job(self):
        return self._selected_job

    async def start_processing(self, selected_job):
        self._selected_job = selected_job
        progress_stages = (
            Progress('Start processing', 0.1), Progress('Processing...', 0.3),
            Progress('Processing some more...', 0.5), Progress("Done", 1)
        )
        for stage in progress_stages:
            await asyncio.sleep(random.randrange(1, 10))
            await self._message_queue.put(stage)
        await self._message_queue.put(None)

    async def get_progress(self):
        progress = await self._message_queue.get()
        if progress is None:
            self._selected_job = None
        self._latest_stage = progress
        return progress

    def latest_stage(self):
        return self._latest_stage

As for the server, I'm using Sanic - a framework similar to flask, but asyncio based. Let's get the uninteresting boilerplate code out of the way first:

import logic
from sanic import Sanic
from sanic.response import stream, html, text
from jinja2 import Environment, PackageLoader
import json
import asyncio

env = Environment(loader=PackageLoader('app', 'templates'))
app = Sanic(__name__)
app.static('/static', 'app/static')

def render_template(html_file, **kwargs):
    template = env.get_template(html_file)
    html_content = template.render(url_for=app.url_for, **kwargs)
    return html(html_content)

Now that we got this out of the way - to the meat:

Currently, the web app supports a single manager, so it's initialized as a singleton, using sanic's event loop


@app.listener('after_server_start')
def create_jobManager(app, loop):
    asyncio.set_event_loop(loop)
    app.job_manager = logic.JobManager(loop)

For the GET method, it renders the job selection screen if no job has been selected, and the job monitoring screen if a job is under processing.

In the case that a job had been selected, the POST method had been used, and in this case the job_manager is told to start the long process with

asyncio.gather(app.job_manager.start_processing(selected_job))

Here's the complete code:

@app.route("/", methods=('GET', 'POST'))
def home(request):
    if request.method == 'GET':
        if app.job_manager.current_job():
            return render_template('process_and_monitor.html', selection=app.job_manager.current_job())
        else:
            return render_template('select_job.html', names=logic.get_names())
    elif not app.job_manager.current_job():
        selected_job = request.form.get('selection', None)
        if selected_job:
            asyncio.gather(app.job_manager.start_processing(selected_job))
            return render_template('process_and_monitor.html', selection=selected_job)
    else:
        return render_template('process_and_monitor.html', selection=app.job_manager.current_job())

Now the only thing left is reporting the progress:

@app.route("/get_progress", stream=True)
async def get_progress(request):
    async def jsonify(response):
        if app.job_manager.latest_stage():
            await response.write(f"data: {json.dumps(app.job_manager.latest_stage()._asdict())}\n\n")
        while True:
            progress = await app.job_manager.get_progress()
            if progress is None:
                return
            await response.write(f"data: {json.dumps(progress._asdict())}\n\n")
    return stream(jsonify, content_type='text/event-stream')

On the client side, the update looks like this:

window.onload = function (e) {
    var source = new EventSource("get_progress");
    source.onmessage = function(event) {
        try {
            const response = JSON.parse(event.data);
            document.getElementById("progress").innerHTML = response.stage;
            document.getElementById("progress_bar").style.width = (response.progress * 100) + "%";
            if (response.progress == 1) {
                document.getElementById("back_button").style.display = "block";
                source.close();
            }
        } catch (error) {
            document.getElementById("progress").innerHTML =  "<h1 class=\"error\">Failed parsing progress</h1>";
            console.log(error);
        }
    }
}

I have omitted a status poll which is triggered by the client every second, to make sure the server is ok (since if it stopped working, the page would never get updated, and there would be no way to know something went wrong), but that code is uninteresting.

This is my first web app, my fist use of asyncio, sanic, jinja, and javascript - any suggestion regarding best practices, readability, pointing out stupid things I did, or things that could be done better using other modules or approaches is welcome.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.