
[Celery](https://docs.celeryq.dev/en/stable/index.html) is probably the most used task runner in the Python ecosystem. Yet, I've found today that configuring it to support a basic scenario for such a critical piece in a software architecture surprisingly difficult and poorly documented.

So, what's the problem? Let's say we have a task taking several minutes to run. Imagine now that we have a server failure, completely independent of our source code: a server restart, a power failure, etc. Thus, the task was stopped in the middle of execution and didn't finished. Fine! Things like that happen often with computers 😏 All we want now is Celery to **execute the task again** after the server restart!

Well, actually, by default: **it won't** 🥹 This is actually an [intended default](https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.acks_late) of Celery. By the way, this is one of the [main painpoint raised by the author of Dramatiq](https://dramatiq.io/motivation.html#id2), a competing library.

Fortunately, there is a way to do it! And this blog post will save you hours of research 🙃

## TL;DR

Here is the configuration you need to apply for this to work:

```py
app = Celery(
    "tasks",
    broker="YOUR_BROKER",
    backend="YOUR_BACKEND",
)
app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_state_db="./celery-state.db",
)
```

Don't go now! I highly recommend you to read the rationale about it 👇

## Rationale

So what those options are doing? Basically, we need to tell Celery:

* To acknowledge the task *after* their completion, not right before (`task_acks_late`).
* To reject the task if the worker is lost or shutdown (`task_reject_on_worker_lost`).
* By default, task rejections are persisted only in memory and thus lost if the server is stopped. So we need to setup a state file (`worker_state_db`).

## The visibility timeout quirk

Still, if you test to manually kill the worker while a long task is running before restarting it, you'll see that the task looks like **stuck and never re-executed**.

Actually, this is **normal** behavior. Why? Because tasks are not re-delivered until they reach the **visibility timeout**. And by default, this timeout is 1 hour. It means that your pending task will be executed again in 1 hour.

If you set this value to 30 seconds, you'll see that your task will get executed again within 30 seconds.

It might be tempting to decrease this visibility timeout and keep a very low value but bear in mind that it'll apply for **every tasks**, not only the crashed ones. If you have a task that takes more than 1 hour to execute, it'll be automatically rescheduled on another worker. That's why it's important to keep it quite high, especially if we have long tasks to run.

## References

* https://docs.celeryq.dev/en/stable/faq.html#faq-acks-late-vs-retry
* https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-acks-late
* https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-reject-on-worker-lost
* https://docs.celeryq.dev/en/stable/userguide/workers.html#persistent-revokes
* https://docs.celeryq.dev/en/stable/getting-started/backends-and-brokers/redis.html#id1


Configure Celery for reliable delivery


In recent years, Python has gained lot of popularity in the data science community, thanks to the vast ecosystem of libraries covering both the scientific aspect, like pandas or scikit-learn, but also the operational aspect, with highly popular web frameworks like Flask or Django. With one unique language, data scientists are able to embed their experiments and results directly in production-ready applications. In 2018, a new challenger blew up the landscape: [FastAPI](https://fastapi.tiangolo.com/). In less than three years, it has become the #3 most-used web framework, right behind Flask and Django. Its ease of use and intuitiveness rapidly made it the go-to tool for data scientists willing to package their work into powerful REST API.

To convince you, we'll show in this article how to create a reliable data ingestion service, with the help of FastAPI, [SQLModel](https://sqlmodel.tiangolo.com/) and [Dramatiq](https://dramatiq.io/). This service should expose a REST API accepting raw documents in a fast and reliable way, able to handle thousands of documents per second. In the background, it should run some prediction algorithms on those documents' content before saving them into a database.

All the source code is [available on GitHub](https://github.com/frankie567/fastapi-dramatiq-data-ingestion).

## Understand the web-queue-worker architecture

Traditionally, REST API endpoints validate user input, perform some operations and save the data into a database. While this pattern is perfectly working in most applications, we could hit a bottleneck in the context of our data ingestion service: with thousands of documents per second, we can't be sure the web server will be able to handle them fast enough without blocking and take the risk of loosing some documents in the process.

That's why we'll consider here what is generally called the **web-queue-worker** architecture. Basically, the whole application is composed of three things:

* A web service, which is our user-facing API that will accept new data and send tasks to a queue;
* A queue, which will store those pending tasks for later;
* A worker service, which will pull new tasks from the queue and handle them in background.

```mermaid
flowchart LR
    subgraph SERVICE[Data ingestion service]
        API[Web service]
        QUEUE[(Queue)]
        WORKER[Worker]
        API -- sends task --> QUEUE
        QUEUE -- pulls task --> WORKER
    end
    WEB{Web}
    WEB --> API
```

By doing this, we are **moving all the heavy logic in a background process**, the worker. Hence, the role of the web service is just to validate input and put a new task in the queue. Usually, queues use very fast and efficient technologies, like RabbitMQ or Redis. Hence, the operation inside the API is very quick, allowing the web server to answer the request rapidly; which is highly desirable if we have lot of requests!

The worker can then process the tasks as they come. Even if there are lot of them, it's not an issue: every pending tasks are safely stored in the queue, so we limit the risk of losing data. If we wish to handle the queue more rapidly, all we have to do is to spawn more worker processes that will pull and handle tasks concurrently.

```mermaid
flowchart LR
    subgraph SERVICE[Data ingestion service]
        API[Web service]
        QUEUE[(Queue)]
        WORKER_1[Worker 1]
        WORKER_2[Worker 2]
        WORKER_3[Worker 3]
        API -- sends task --> QUEUE
        QUEUE -- pulls task --> WORKER_1
        QUEUE -- pulls task --> WORKER_2
        QUEUE -- pulls task --> WORKER_3
    end
    WEB{Web}
    WEB --> API
```

This architecture may sound complex to implement but actually, thanks to FastAPI and Dramatiq, you'll see that we can do it very quickly. Here is the plan:

* Web service will be implemented with FastAPI;
* Worker and task management will be handled by Dramatiq;
* The queue data storage (we say **broker**) will be Redis;
* The actual documents data will be stored on a SQL database. The communication with this database will be handled by SQLModel.

```mermaid
flowchart LR
    subgraph SERVICE[Data ingestion service]
        API[FastAPI]
        QUEUE[(Redis)]
        WORKER[Dramatiq]
        DB[(SQL database)]
        API -- sends task --> QUEUE
        QUEUE -- pulls task --> WORKER
        DB -- reads data --> API
        WORKER <-- reads/writes data --> DB
    end
    WEB{Web}
    WEB --> API
```

## Define the SQLModel models

Before anything else, let's define the data models for our application. As we said, we'll use the SQLModel library to help us with that. SQLModel is a brand-new library by the creator of FastAPI, which combines the power of Pydantic models, already at the heart of FastAPI, and SQLAlchemy, the unavoidable SQL library for Python. It allows us to define our SQL data with standard Python classes and type annotations. Let's see it in action:

```py
import uuid
from datetime import datetime
from typing import Optional

from pydantic import UUID4
from sqlmodel import Field, SQLModel


class DocumentInput(SQLModel):
    id: UUID4 = Field(default_factory=uuid.uuid4, primary_key=True)
    content: str


class Document(DocumentInput, table=True):
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
    category: Optional[str] = Field(default=None)

```

We define two models: `DocumentInput` and `Document`.

The first one will be the data structure accepted by our REST API. The great thing about FastAPI and Pydantic models is that data validation is already covered: if a property is missing or doesn't have the right type, FastAPI will automatically generate a proper error response. Here, we define two properties: `id` and `content`, which will bear the raw content of the document.

We chose to use UUID instead of a classic auto-increment number as primary key. It's an interesting pattern in this context because the client can set the id by itself; which may be useful if we need to send an updated version of the same document. If we relied on an auto-increment identifier, we would need first to make a request to retrieve it or save it somewhere else.

The second one, `Document`, is the actual data structure for our SQL database (notice the `table=True` argument). It inherits from `DocumentInput` to which we have added creation/update dates and a **category field**. This category will be **determined automatically during data ingestion** thanks to a machine learning model.

## Implement the worker

Let's now have a look at the worker. Basically, it's a Python module containing all the tasks that will be handled by Dramatiq. We also initialize here the various parameters for Dramatiq, especially the broker. Here is how it looks like:

```py
from urllib.parse import urlparse
from datetime import datetime

import dramatiq
from dramatiq.brokers.redis import RedisBroker
from sqlmodel import Session

from app.db import engine
from app.models import Document, DocumentInput
from app.predict import CategoryPrediction
from app.settings import settings

redis_parameters = urlparse(settings.redis_url)
redis_broker = RedisBroker(
    host=redis_parameters.hostname,
    port=redis_parameters.port,
    username=redis_parameters.username,
    password=redis_parameters.password,
    # Heroku Redis with TLS use self-signed certs, so we need to tinker a bit
    ssl=redis_parameters.scheme == "rediss",
    ssl_cert_reqs=None,
)
dramatiq.set_broker(redis_broker)


category_prediction = CategoryPrediction()


@dramatiq.actor
def ingest_document(document_json: str):
    document = DocumentInput.parse_raw(document_json)
    with Session(engine) as session:
        document_db = session.get(Document, document.id)
        if document_db is None:
            document_db = Document(**document.dict())
        else:
            document_dict = document.dict(exclude_unset=True)
            for key, value in document_dict.items():
                setattr(document_db, key, value)

        document_db.category = category_prediction.predict(document_db.content)
        document_db.updated_at = datetime.utcnow()

        session.add(document_db)
        session.commit()
```

The first thing to notice is that we set the broker to be Redis by specifying the Redis connection parameters.

Then, we have our task definition. As you can see, it's a standard Python function upon which we added the `@dramatiq.actor` decorator. It receives in argument a JSON representation of a `DocumentInput`. Indeed, we can't pass directly Python objects because Dramatiq needs to serialize them before saving into the broker. That's why we first deserialize it back again into a proper `DocumentInput` object.

Next, we first check if we don't already have a document with the same id in the database. This way, we can update it or create a new one if needed. `session` and `engine` are SQLModel tools used to communicate with the database. You can read more about how it works in [the official documentation](https://sqlmodel.tiangolo.com/tutorial/one/#select-by-id-with-get).

### Add a machine learning model to predict category

Actually, the most important step in this function is the category prediction. You can see that we use a `CategoryPrediction` class for this purpose. This utility class allows us to load a model that have been trained previously in scikit-learn. In this example, this is a simple [Multinomial Naive Bayes classifier](https://scikit-learn.org/stable/modules/naive_bayes.html#multinomial-naive-bayes) trained on [the 20 newgroups text dataset](https://scikit-learn.org/stable/datasets/real_world.html#newsgroups-dataset).

The purpose here is not to provide a very powerful model for classifying documents but rather to show you how you can integrate a scikit-learn estimator into a data ingestion service. By putting this operation in a worker process, like here, we can imagine even very expensive algorithms to apply on our documents: it won't break our system.

Here is how this class looks like:

```py
import os
from typing import List, Optional, Tuple, cast

import joblib
from sklearn.pipeline import Pipeline


class CategoryPrediction:
    model: Optional[Pipeline] = None
    targets: Optional[List[str]] = None

    def load_model(self):
        """Loads the model"""
        model_file = os.path.join(os.path.dirname(__file__), "newsgroups_model.joblib")
        loaded_model: Tuple[Pipeline, List[str]] = joblib.load(model_file)
        model, targets = loaded_model
        self.model = model
        self.targets = targets

    def predict(self, input: str) -> str:
        """Runs a prediction"""
        if self.model is None or self.targets is None:
            self.load_model()
        model = cast(Pipeline, self.model)
        targets = cast(List[str], self.targets)
        prediction = model.predict([input])
        return targets[prediction[0]]
```

This class has two methods: the first one to load a dumped model with [Joblib](https://joblib.readthedocs.io/en/latest/). This library allows us to dump arbitrary Python objects in a file on disk, like a trained scikit-learn estimator. This way, we can load it again on our production service and use it right away to perform predictions.

The second method actually perfoms a prediction on an input string. Since the loaded object is a true scikit-learn estimator instance, we can use it like usual. Notice that we load the model *lazily*: it's not loaded until we perform our first prediction. Since loading the model can be quite expensive, we don't want the startup of our worker process to be slowed down by this operation.

## Implement the REST API

It's now time to implement the REST API that will accept new documents. By using the models we defined before, you'll see that FastAPI allows us to implement an endpoint with proper input validation in no time. Here is how the *whole* application looks like:

```py
from fastapi import FastAPI, HTTPException, status
from pydantic import UUID4
from sqlmodel import Session

from app import models
from app.db import create_db_and_tables, engine
from app.worker import ingest_document

app = FastAPI()


@app.post(
    "/documents",
    status_code=status.HTTP_202_ACCEPTED,
    response_model=models.DocumentInput,
)
def add_document(document: models.DocumentInput):
    ingest_document.send(document.json())
    return document
```

Yes. You don't need any more line to make your API working. That's the beauty of FastAPI: just by using the proper type hint in your function argument, the framework will validate that the request payload should contain a JSON object structured following a `DocumentInput`.

All we have to do then is to import the `ingest_document` task from our worker module and `send` a new task in the queue. Notice that we serialize the object in JSON before. At that point, the document is sent in the Redis queue and will wait to be handled by the worker.

And that's all! By keeping it very simple, our endpoint will be able to handle thousands of requests without any problem.

## Deploy on Heroku

Our service is ready! It would be nice if we could try it, right? Actually, we can deploy it for free in just a few minutes! We'll use [Heroku](https://www.heroku.com/), one of the leading serverless platforms. The main benefit of those kind of platforms is that you don't have to manage a server yourself: you just configure your application and send the source code. The platform takes care of deploying it and make it available to the whole world!

The first step before deploying to Heroku is to create a `Procfile`. This is a special file for Heroku that defines what processes you need and how to run them. It looks like this:

```
web: uvicorn app.api:app --host 0.0.0.0 --port $PORT
worker: dramatiq -p 4 -t 4 app.worker
```

As we said before, we have two main processes:

* The web process, i.e. our FastAPI application. On Heroku, a process named `web` means it's a process that needs to be accessible through the web and routed accordingly. On the right, we have the command the platform should run to start the FastAPI application. [Uvicorn](https://www.uvicorn.org/) is a web-server able to work with ASGI applications, the standard Python protocol for asynchronous web applications like FastAPI. The first parameter is the import path towards our FastAPI app variable.
* The worker process, which is run by Dramatiq. Simply, we use the `dramatiq` command and pass the import path of our worker module.

What about Redis and the SQL database? In Heroku, you can attach external services that will be entirely managed by the platform or third-party providers. Those are the add-ons, and Heroku proposes ones to have Redis and PostgreSQL servers. We'll see how to add them in a minute.

Heroku applications can be managed through their web interface or their CLI tool. We'll show you how to deploy the application with the CLI tool. First of all, create a new Heroku account and [install the CLI](https://devcenter.heroku.com/articles/heroku-cli).

Once done, the first thing to do is to create a new app:

```bash
heroku apps:create fastapi-data-ingestion --region=eu
```

This will create an app named `fastapi-data-ingestion` in the European data center.

Heroku stores and updates your application source code through a standard Git repository. To do this, it sets a special remote in your Git repository: when you push on this remote, Heroku automatically deploys a new version of your application. The following command will setup the remote URL for your app in your current Git repository:

```bash
heroku git:remote -a fastapi-data-ingestion
```

Great! Let's now add a Redis and a PostgreSQL add-on on the free plan:

```bash
heroku addons:create heroku-redis:hobby-dev
heroku addons:create heroku-postgresql:hobby-dev
```

The nice thing about add-ons is that they **automatically set environment variables** with the right authentication information. Those environment variables are available in your application and can be used directly to connect to those services. You can check the defined environment variables for your application with the following command:

```bash
heroku config
```

You'll get an output like this:

```
=== fastapi-data-ingestion Config Vars
DATABASE_URL:  postgres://********************************
REDIS_TLS_URL: rediss://********************************
REDIS_URL:     redis://********************************
```

Everything is ready! All we have to do now is to push our code to the Heroku remote to start a deployment. Let's go:

```bash
git push heroku main
```

The deployment will take a few minutes. You'll see in the output all the operations Heroku performs, especially the installation of Python dependencies: don't forget to define your `requirements.txt`!

Once done, there is one last thing to do. By default, Heroku doesn't start processes other than `web`. The following command will start the worker process:

```bash
heroku ps:scale worker=1
```

That's it! Our data ingestion service is up and running! Wanna try it? Let's send a document into our REST API:

```bash
curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"content":"computers are awesome"}' \
  https://fastapi-data-ingestion.herokuapp.com/documents
```

You'll get the following result:

```json
{"id": "179611c1-a124-4403-905d-0f0dbe7e6d28", "content": "computers are awesome"}
```

An id has been generated for our new document. How to be sure that it has been properly handled by our worker process? We've implemented another API endpoint to retrieve a saved document. Let's try it:

```bash
curl https://fastapi-data-ingestion.herokuapp.com/documents/179611c1-a124-4403-905d-0f0dbe7e6d28
```

Which should give you:

```json
{
    "id": "179611c1-a124-4403-905d-0f0dbe7e6d28",
    "created_at": "2021-09-01T12:10:51.738076",
    "updated_at": "2021-09-01T12:10:52.085624",
    "category": "comp.sys.mac.hardware",
    "content": "computers are awesome"
}
```

Awesome! Our document *is* in the database and a category have been predicted for it.

## Final thoughts

Congratulations! You now know how to implement the web-queue-worker architecture in Python using FastAPI, Dramatiq and SQLModel. As you've seen, those technologies help you implement this pattern in no time thanks to an easy-to-use and readable syntax. With this kind of service, you'll be able to handle a large stream of documents and apply expensive machine learning algorithms, like here with a scikit-learn estimator, without compromising the stability of the application or the data integrity.

I hope you liked this quick walkthrough and that it gave you ideas for your next projects. If you wish to go further and master FastAPI to build efficient and reliable data science applications, you can check my book **Building Data Science Applications with FastAPI**, which is available in e-book and paperback on [Amazon.com](https://www.amazon.com/gp/product/B09926TFQ5). Here is what you'll learn:

* The concepts of the FastAPI framework, including aspects relating to asynchronous programming, type hinting, and dependency injection
* Develop efficient RESTful APIs for data science with modern Python
* Build, test, and deploy high performing data science and machine learning systems with FastAPI

[![Building Data Science Applications with FastAPI book](/posts/images/create-deploy-reliable-data-ingestion-service-fastapi-sqlmodel-dramatiq/fastapi-data-science-book-cover.png)](https://www.amazon.com/gp/product/B09926TFQ5)

## Watch the video

<iframe width="560" height="315" src="https://www.youtube.com/embed/zC670X5_zhw" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe>

See you soon 👋


Create and deploy a reliable data ingestion service with FastAPI, SQLModel and Dramatiq


Asynchronous programming have become very popular in the recent years, NodeJS being the locomotive of this trend, highly encouraging the async style and backed by impressive benchmarks.

In recent months, other languages have started to catch up, implementing their own flavor of asynchronous style. Among them, of course, almighty Python (*did I tell you I love Python?*).

But before talking about the Python ecosystem, let's have a quick overview of how asynchronous programming actually works.

## How does asynchronous work?

The main motivation behind asynchronous style can be summed up in one simple statement: I/O is painfully slow. Reading from disk, networks requests, are million times slower than reading from RAM and processing instructions. Traditionally, in a synchronous program, you would do things like this:

```py
with open("file.txt") as f:
    data = f.read()
# The program will block here until the data has been read
print(data)
```

We see that the program will block until we retrieved the data from the disk ; and, as we said, this can be long. 99% percent of the execution time of the program is just to wait for the disk. It's usually not a big deal for simple scripts like this because you probably don't have to perform other operations meanwhile.

However, in more complex cases, this is a waste of resources that could have been used to perform **other things**. For example, think of a web server: if a first user makes a request that needs to block the program for 10 seconds, a second user making a request meanwhile would have to wait for the first one to finish before they get answered. We can simulate this behaviour quite easily with flask and gunicorn:

```py
# app.py
import time
from flask import Flask, request

app = Flask(__name__)

@app.route("/slow")
def slow():
    time.sleep(10)
    return "That was slow!"

@app.route("/fast")
def fast():
    return "That was fast!"
```

```sh
gunicorn --workers=1 app:app
```

If you hit the `/slow` endpoint and right after the `/fast` endpoint, you'll see that you get the `/fast` response only when `/slow` has answered (after 10 seconds have passed).

To circumvent this, web servers usually spawn several threads (notice that we purposely started gunicorn with 1 worker) to be able to serve several users at a time, even if one request takes time to complete.

Solution adopted in asynchronous style doesn't involve threads, as they come with their own difficulties. Asynchronous frameworks implement a concept you probably have already heard of: **event loops**.

An event loop basically is the conductor of your program: at each iteration, it'll check for the completion of the tasks, yield the result if it's done, and go on to the next task. As a developer, your role is thus to send tasks to the event loop, knowing that you'll eventually get their result. Let's rebuild our previous example asynchronously with FastAPI and Uvicorn:

```py
# app.py
import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/slow")
async def slow():
    await asyncio.sleep(10)
    return "That was slow!"

@app.get("/fast")
async def fast():
    return "That was fast!"
```

```sh
uvicorn --workers=1 app:app
```

If you spawn the two requests at the same time like earlier, you'll see that `/fast` has answered immediately, not waiting for `/slow` to finish: the event loop put the `/slow` request in its queue, and moved on to the next one.

Notice that we traded `time.sleep` for `asyncio.sleep`. This is important because the event loop can only manage asynchronous tasks, that give back control to their caller. If you keep `time.sleep`, you'll experience the same behaviour as in the previous example: the thread will be blocked because it didn't yielded to the event loop.

Fine. But how does that solve our I/O problem? If we want to work with our event loop, I/O operations have to work asynchronously. To do this, we leverage an interesting feature of the operating systems: the select and poll calls (and derivatives). Those allow to ask the OS to ping us when something happened on a file or a socket (new data to read, ready to write data...). By using them, we allow the event loop to put the I/O calls in background and handle them once the OS notified us.

That was a very simplified overview of how asynchronous is implemented. If you wish to have a more complete and correct view of what is going on under the hood, be sure to check the references.

## The Python async ecosystem

`asyncio` module is included since Python 3.4 (2014!), and the async/await syntax since Python 3.5. Both have been greatly refined in 3.6 and 3.7; making the use of async style very easy. Logically, the ecosystem has grown since: there are now solid libraries ready for production use. We will just cite a few of them:

* [Starlette](https://www.starlette.io/), an impressively fast and clean framework to build web applications ;
* Its offspring, [FastAPI](https://fastapi.tiangolo.com/), which builds upon Starlette and [Pydantic](https://pydantic-docs.helpmanual.io/) to propose a very efficient developer experience to build API ;
* [HTTPX](https://github.com/encode/httpx), an async-compatible HTTP client with a requests-like API.
* [Tortoise ORM](https://tortoise-orm.readthedocs.io/en/latest/), a Django-like ORM.
* [Motor](https://motor.readthedocs.io/en/stable/index.html), an official MongoDB client.

All those tools allow to build performant web services leveraging the power of the asynchronous paradigm.

## Conclusion

Asynchronous programming is a very interesting pattern that can greatly improve performance in some contexts, especially in web servers. While it has been the privilege of NodeJS for quite some time, other languages are now catching up and it's great to see that Python community is becoming more and more involved in this trend.

As a Python and FastAPI enthusiast, I'm glad to contribute to this ecosystem by releasing open-source libraries like [FastAPI Users](https://github.com/frankie567/fastapi-users) or [HTTPX OAuth](https://github.com/frankie567/httpx-oauth).

## References

* https://en.wikipedia.org/wiki/Asynchronous_I/O
* https://en.wikipedia.org/wiki/Async/await
* https://docs.python.org/3/library/selectors.html
* https://trio.readthedocs.io/en/stable/tutorial.html
* https://trio.readthedocs.io/en/stable/design.html
* https://stackoverflow.com/questions/49005651/how-does-asyncio-actually-work
* http://blog.mixu.net/2011/02/01/understanding-the-node-js-event-loop/
* https://snarky.ca/how-the-heck-does-async-await-work-in-python-3-5/
* https://jvns.ca/blog/2017/06/03/async-io-on-linux--select--poll--and-epoll/


Asynchronous programming is now a thing in Python


Most services we are using nowadays are web-based, and it’s easy to understand why: for the developers, it’s easy to develop, maintain and update. The users have nothing to install and can start to use the product in just a few clicks.

Thanks to the evolution of web technologies these last few years, we can indeed have interactive interfaces that have nothing to envy to good-old desktop applications.

However, in some use cases, a desktop app is still needed. Think for example about services that need to access your local files to synchronize them, like Dropbox or Wuha. It could also be an app that needs to access the local devices, like your GPS in order to update the map.

For this to work, we need them to develop an app with a desktop interface, different and separated from our web one. Painful, hard to maintain, very different from web technologies (we’ll probably need a dedicated developer for that).

**What if we could just have a daemon process giving us access to the local computer that we could control from our web service?**

## Hello, WebSockets!

The problem here is to make the desktop process able to send and receive data to and from our web service. Traditionally, a client makes requests to a server synchronously: its posts the data of a form, and gets a response in return.

![Desktop making HTTP requests to server](/posts/images/control-desktop-app-through-websockets/desktop-server.png "Desktop making HTTP requests to server")

However, how can the server **send** requests to the desktop ? It can’t. Unless we also make the desktop app a server itself. It comes with loads of issues: firewall rules? Proxying? Security? We need a better alternative.

Here comes **WebSockets**: it’s a web standard protocol aiming at opening a full-duplex communication channel between a client and a server. In other words, it opens a tunnel in which correspondents can both **send and receive messages**.

![WebSocket between server and desktop](/posts/images/control-desktop-app-through-websockets/websocket.png "WebSocket between server and desktop")

So, now, what about the **web interface** ? Well, we can just open a WebSocket between the browser and the server ; then, the server will forward the messages from the browser to the desktop app and conversely.

![Server making the bridge between the WebSockets of web and desktop](/posts/images/control-desktop-app-through-websockets/web-server-client.png "Server making the bridge between the WebSockets of web and desktop")

Enough talking, now, let’s code!

## Thank you, NodeJS, but we’ll take it from here

![NodeJS getting bitten by a Python](/posts/images/control-desktop-app-through-websockets/rogue-nodejs.jpeg)

While asynchronous paradigm has been the prerogative of NodeJS for a while, many other languages have now implemented an asynchronous API ; like Python, a *real* language we **all** love (if you don’t, well, you **should** love it).

Why I talk about asynchronous programming? Because we’ll use the Python [Starlette framework](https://github.com/encode/starlette) (from the creator of Django REST framework) which is an API framework leveraging the power of `asyncio`. It also implements an API to create a WebSocket server.

### Prototype

We’ll implement a very simple prototype:

* The server will match a web client and a desktop client through a client_id, and broadcast the messages between them.
* The desktop will report the computer CPU usage each second and make a beep when it receives the appropriate message.
* The web will display the CPU usage it receives and will propose a button to send the beep message.

We’ll go through some parts of the code. The entire implementation is available on GitHub. Try it yourself!

<iframe id="ghcard-frankie567-1" frameborder="0" scrolling="0" allowtransparency="true" src="//cdn.jsdelivr.net/github-cards/1.0.2/cards/default.html?user=frankie567&amp;identity=ghcard-frankie567-1&amp;repo=remote-desktop-control" width="100%" height="150"></iframe>

### Server

The server just accepts WebSockets connections and matches a web client and a desktop client to broadcast the messages between them.

```py
@app.websocket_route('/ws')
async def websocket_endpoint(websocket):
    await websocket.accept()

    # "Authentication" message
    message = await receive_json(websocket)
    client_mode = message['client_mode']
    client_id = message['client_id']
    websockets[client_mode][client_id] = websocket
```

As you can see, serving a WebSocket is a piece of cake with Starlette. After having established the connection, we expect the client to send us an “authentication” message so that we can match it with the other client.

Obviously, in a real-world application, we would have a proper authentication with a token.

We keep the websocket in memory so that we can broadcast messages into it:

```py
try:
    # Broadcast it to the mirror client
    await websockets[mirror_mode][client_id].send_text(
        json.dumps(message)
    )
except KeyError:
    logger.debug(
        f'Client {client_id}[{mirror_mode}] not connected'
    )
```

### Desktop

The desktop also uses the `asyncio` API to report the CPU usage every seconds and reacts to the beep messages.

```py
async def handler(uri, client_id):
    async with websockets.connect(uri) as websocket:
        message = {
            'event': 'authentication',
            'client_id': client_id,
            'client_mode': 'desktop'
        }
        await websocket.send(json_to_payload(message))

        consumer_task = asyncio.ensure_future(
            consumer_handler(websocket))
        producer_task = asyncio.ensure_future(
            cpu_usage_reporter(websocket))
        done, pending = await asyncio.wait(
            [consumer_task, producer_task],
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in pending:
            task.cancel()
```

### Web

Connecting to WebSockets is very easy in browsers. Just a few lines of JavaScript is enough.

### Let’s see it in action

<iframe width="560" height="315" src="https://www.youtube.com/embed/exkqQUW9O_A" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe>

## One interface to rule them all

We have demonstrated an easy way to create a communication channel between a desktop process and a web-based interface by using standard WebSockets.

Approaches like this allow us to **access the local resources** of the computer while **keeping the user in the same, unique interface**. Less confusion, less code to maintain.


Control a desktop app through web with WebSockets


> In 2013, during my first year of Machine Learning and Data Mining Master’s Degree, I took a very interesting course about Music Information Retrieval ; a field between musicology, physics and machine learning aiming at extracting information from musical data.
>
> We were asked to write a small report on a musical genre of our choice and explain how it could be interesting for the research in this field.
>
> My interest for History made me to work on a forgotten genre, the air de cour, a very popular music genre at the French court and parisian salons in the XVIIth century, essentially vocal and sometimes accompanied by a lute.
>
> Here is this report, mined from the bottom of one of my folders, which is, I believe, not without interest.

<iframe width="100%" height="166" scrolling="no" frameborder="no" allow="autoplay" src="https://w.soundcloud.com/player/?url=https%3A//api.soundcloud.com/tracks/110385172&color=%23ff5500&auto_play=false&hide_related=false&show_comments=true&show_user=true&show_reposts=false&show_teaser=true"></iframe>

*An air de cour, brilliantly interpreted by the [Ensemble Faenza](http://www.faenza.fr/), to get into the mood. 🎵*

## Abstract

In this report, we will study a specific musical genre, who were very popular in France at the royal court, between the late 16th century and early 17th, the “Air de cour”. Nearly forgotten today, we will see its features and the difficulties one may encounter to read and process these old compositions, melting scores in mensural notations, lyrics in old-fashioned French and lute tablatures.

## Introduction

At the end of the 16th century, France is marked by the wars of religions between Catholics and Protestants. At the end of the Renaissance, period where Europe changed completely its vision of the world, we see the emergence of a new artistic movement, trying to show the exuberance and the grandeur, the Baroque, through all the arts, and of course in music. It is in this context that appear the “Airs de cour”, a very specific musical genre, nearly forgotten today.

## Air de cour

Adrian le Roy (c. 1520-1598), a french musician of the Renaissance, was the first to use the term “Air de cour” with the publication of his collection of songs “Airs de cour miz sur le luth”[^1] in 1571.[^2] This genre were very popular in France at the royal court, and more widely with the Parisian aristocracy and the “salons” until the mid-17th, making “Air de cour” a musical style placed at the border between Renaissance and Baroque periods.

“Air de cour” is essentially a vocal music genre. With time, it successively passed from solo voice to multiple voices, and again solo. It could come with no accompaniment but, usually, it is accompanied by a lute ; therefore, composers write them with the vocal melody, the lyrics in French and the lute accompaniment, as we will see.

Many composers from the late 16th century to early 17th, such as Étienne Moulinié or Bénigne de Bacilly, composed some “airs de cour”.

![Charles Mouton, a famous luth player, by Gérard Edelinck, 1692](/posts/images/air-de-cour-interest-music-information-retrieval/charles-mouton.jpg "Figure 1: Charles Mouton, a famous luth player, by Gérard Edelinck, 1692")

### Musical approach

An “air de cour” is composed of several stanzas, each one sung on the same melody and split in two parts, the second one being repeated once more, playing the role of a chorus. The metric is quite unusual and may appear irregular for us today. Indeed, “airs de cour” were heavily influenced by the principle of “musique mesurée”[^3], a style trying to adapt the rhythm of the melody to the verse ; longer syllabes set to longer notes and shorter syllabes to shorter notes.

There is today few artists who recorded these musical pieces. We can however mention the work of Suzie Le Blanc[^4] on Moulinié’s compositions or that of Claudine Ansermet[^5] on Bacilly’s work. While listening, we clearly hear the two influences of the Renaissance, with the light lute accompaniment, and the Baroque, with the way of declamating the lyrics that will influence the french opera later, especially the french lyric tragedy of Lully.

## Study of musical scores

As few studies were lead on this particular style, the best way to understand it is to study in more detail some scores of this time. Fortunately, some of this scores are digitized and freely available to download on the International Music Score Library Project (IMSLP), a project who tries to scan old scores on public domain and publish them to the web. We chose here to focus on the “Airs de cour miz sur le luth” by Adrien le Roy[^6] and on the first book of “Airs de cour” by Étienne Moulinié[^7].

### Structure

The presentation and structure of a piece is quite similar in both books. We find the three essential elements : the melody score, the lyrics and the lute part. It is important to notice that the lute is written in the form of a tablature and not with a pitch representation. We don’t really know why, apart from the fact that this representation is quite usual for string instruments like the lute or the guitar.
Concerning the lyrics, we can say that, even if they are not in what we call usually “Old French”, they are in a old-fashioned French and with some typographic particularities ; but they are still understandable without particular knowledge.

### Musical notation

As we stated before, “airs de cour” are at the edge of the Renaissance and the Baroque periods. This is why, in Le Roy’s book, we have a mensural notation8 whereas in Moulinié’s one, we already have a form of our modern notation, as we can see in Figure 2a and Figure 2b.

![Extract of a score from Adrien Le Roy (1571)](/posts/images/air-de-cour-interest-music-information-retrieval/leroy_score.png "Figure 2a: Extract of a score from Adrien Le Roy (1571)")

![Extract of a score from Étienne Moulinié (1624)](/posts/images/air-de-cour-interest-music-information-retrieval/moulinie_score.png "Figure 2b: Extract of a score from Étienne Moulinié (1624)")

We say “a form” because, as you can see in Moulinié’s score, it is still melted with old-fashioned notations. The G clef, for instance, is quite different from its modern form.

Moreover, it seems that, in both books, metric is written with the mensural notation. As we can see in Figure 3 and Figure 4, they use unusual signs. This needs some explanations. This system is much more complex than the modern one. Indeed, it allowed to define different divisions for two following note types. For example, the division of a Breve into a Semibreve (our modern Whole), called Tempus, could be different from the division of a Semibreve into Minims (our modern Half), called Prolatio. Furthermore, each of these levels could be divided in three (“Perfect” or “Maior” ) or in two (“Imperfect” or “Minor” ). Finally, another notation, a vertical stroke through the previous signs, represents a division by two of all temporal values.

![Detail on a metric notation from Adrien Le Roy (1571)](/posts/images/air-de-cour-interest-music-information-retrieval/leroy_metric.png "Figure 3: Detail on a metric notation from Adrien Le Roy (1571)")

![Detail on a metric notation from Étienne Moulinié (1624)](/posts/images/air-de-cour-interest-music-information-retrieval/moulinie_metric.png "Figure 4: Detail on a metric notation from Étienne Moulinié (1624)")

So, as we can see in Figure 4, we can say that, in this score, a Breve is divided in two Semibreves and Semibreves in two Minims. This corresponds roughly to a meter of 2/4 in our modern notation.

But this doesn’t explain the “3” that we see in Figure 3, and that we find throughout the two books. In this scores, the metric could be changed during the composition : either by inserting one of the previous signs we saw, or by inserting a numerical value, like this “3” that means that the length of the notes must be divided by three.

## Research interest

The “airs de cour” seems to have had little attention from the scientific community. The reasons are multiples: complete desuetude of the genre, very specific in its form (only vocal with one instrument accompanying), lack of reputation of its principal composers, lack of scores… However, thanks to the International Music Score Library Project, we can find some of this scores freely in PDF format, of variable quality though.

Today’s Music Information Retrieval’s system could help to fully digitize all this compositions and save them from a complete oblivion: technologies like Optical Music Recognition (OMR) could be used to detect and transcribe the scores. A research project could try to establish a system solving all the challenging issues we would encounter with “airs de cour” scores :

* Recognize mensural notation, or worse, semi-modern notations like Moulinié’s one ;
* Transcribe such recognized mensural notation in modern system ;
* Process the lute tablature ;
* Recognize and process the lyrics in an old-fashioned French with all its grammatical and typographical issues ; involving Optical Character Recognition technologies, and maybe even Machine Translation’s ones.

---

[^1]: Adrien Le Roy. Livre d’airs de cour miz sur le luth. http://imslp.org/wiki/Livre_d%27airs_de_cour_miz_sur_le_luth_(Le_Roy,_Adrien), 1571.
[^2]: Wikipedia. Air de cour — Wikipedia, the free encyclopedia. https://en.wikipedia.org/wiki/Air_de_cour, October 2013.
[^3]: Wikipedia. Musique mesurée — Wikipedia, the free encyclopedia. https://fr.wikipedia.org/wiki/Musique_mesur%C3%A9e_%C3%A0_l%27antique, October 2013.
[^4]: Suzie Le Blanc. Moulinie: Airs with lute tablature (first book, 1624), 2008.
[^5]: Paolo Ansermet, Claudine Cherici. Benigne de bacilly : L’art de bien chanter, 2006.
[^6]: Adrien Le Roy. Livre d’airs de cour miz sur le luth. http://imslp.org/wiki/Livre_d%27airs_de_cour_miz_sur_le_luth_(Le_Roy,_Adrien), 1571.
[^7]: Étienne Moulinié. Airs de cour premier livre. http://imslp.org/wiki/Airs_de_cour_(Moulini%C3%A9,_%C3%89tienne), 1624.
[^8]: Wikipedia. Mensural notation — Wikipedia, the free encyclopedia. https://en.wikipedia.org/wiki/Mensural_notation, October 2013.


Experiments, thoughts and stories about my work

Configure Celery for reliable delivery

Create and deploy a reliable data ingestion service with FastAPI, SQLModel and Dramatiq

Asynchronous programming is now a thing in Python

Control a desktop app through web with WebSockets

Air de cour : interest for Music Information Retrieval