
When programming, you probably often have, like me, those *Aha!* moments: when things that were not really clear or obvious in your mind suddenly start to make sense. This week, I had one of these... Even if it turned out to also be a *Doh!* moment 😰

This time, it was brought to me by [SQLAlchemy](https://www.sqlalchemy.org/). This Python library is the de-facto standard for working with SQL databases. It's *super* powerful... But also hard to tame.

So let's review a code snippet, typical of what I was doing when working with SQLAlchemy. Let's say we have [SQLAlchemy ORM models](https://docs.sqlalchemy.org/en/20/orm/mapping_styles.html#orm-declarative-mapping) called `User` and `Post` and a ready-to-use [engine](https://docs.sqlalchemy.org/en/20/core/connections.html#sqlalchemy.engine.Engine).

```py
def create_user(session):
	user = User(email="king.arthur@camelot.bt")
	session.add(user)
	session.commit()
	return user

def create_post(session, user):
	post = Post(title="SQLAlchemy: stop calling `session.commit()`", user=user)
	session.add(post)
	session.commit()
	return post

def main():
	with Session(engine) as session:
		user = create_user(session)
		post = create_post(session, user)
```

If that looks okay to you, then read this post in full. Because you're doing it **wrong**.

# `session.commit()` is not your friend

The core of SQLAlchemy usage is the [`Session`](https://docs.sqlalchemy.org/en/20/orm/session_api.html#sqlalchemy.orm.Session) object. Under the hood, besides maintaining ORM objects state, it also manages an **SQL transaction**. It means that you can safely perform queries: if something goes wrong along the way, you can always **rollback** to the previous state, leaving the DB untouched...

... unless you call `session.commit()`.

This method does two things:

1. It flushes all the pending changes to the database (objects creation, update, deletion, etc.).
2. It `COMMIT` the SQL transaction, as its name rightfully suggest: the changes are definitely written to the DB and there is no way back.

Meaning that, if you need to `ROLLBACK`, that's too late. **Even if you are in a nested transaction**. What does it mean if we go back to our example?

Let's say something goes wrong during `create_post`: an API call fails, a bug (😱) occurs, the server shuts down abruptly... The result is that you'll have a user in your database, but not the post; meaning your database is an incoherent state. If you want to retry the task, you'll likely have issues since they are already half of the objects there.

So, what should our example look like to avoid this?

```py
def create_user(session):
	user = User(email="king.arthur@camelot.bt")
	session.add(user)
	return user

def create_post(session, user):
	post = Post(title="SQLAlchemy: stop calling `session.commit()`", user=user)
	session.add(post)
	return post

def main():
	with Session(engine) as session:
		user = create_user(session)
		post = create_post(session, user)
        session.commit()
```

That's the key takeaway of this post. Write it on a post-it. Stick it on your screen:

<p style="text-align:center;font-size: 1.5em;font-weight:bold;color:red">Call session.commit() only once at the end</p>

Said another way, if you have `session.commit()` calls in your business logic, it's a **code smell**.

Generally speaking, `session.commit()` should only appear in "lifecycle" logic, which is managed by your framework or global application scope: a FastAPI dependency, a middleware, a context manager, etc.

# `session.flush()` is probably what you want

At this point, you probably want to ask: *"Ok, but what if I need to execute queries along the way? Or if I want to trigger the database integrity checks or defaults?"* That's indeed a perfectly legitimate concern.

In a complex program, we might create dozens of object in one operation and perform a `SELECT` query, expecting our new data to be available. That's probably why you called `session.commit()` at first.

What you want is actually `session.flush()`. Basically, it's only the first step of the `session.commit()` operation: it sends pending operation to the database, but the transaction is **not** committed. So, it allows us to do this kind of things:

```py
def handle_payment(session, account, amount, fees):
    payment_transaction = Transaction(amount=amount, account=account)
    fees_transaction = Transaction(amount=-fees, account=account)

    session.add(payment_transaction)
    session.add(fees_transaction)

    session.flush()
    # payment_transaction and fees_transaction are flushed to the DB
    #but we're still in the transaction

    account_balance_statement = select(sum(Transaction.amount)).where(
        Transaction.account == account
    )
    account_balance = session.execute(account_balance_statement).scalar()

    account.balance = account_balance
    session.add(account)
```

Here, we are able to update the account balance knowing that the new transactions will be available to the `SELECT SUM...` statement; since we flushed right before. But we'll still have the chance to **rollback** if something goes wrong.

## Autoflush has your back

Actually, in our example above, calling `session.flush()` is not even needed. By default, whenever you `execute` a query, SQLAlchemy will automatically flush pending changes. That's `autoflush`. Most of the time, you probably don't need to worry about it. Don't disable it.

# Conclusion

As I mentioned in introduction, it truly was an *Aha!* moment for me, and now that I re-read this post, it feels so obvious.

And it's also a *Doh!* moment for two reasons:

1. I have to revisit lot of things on the codebases I work on (including Polar).
2. This is actually very well explained in SQLAlchemy documentation, but I missed it: https://docs.sqlalchemy.org/en/20/orm/session_basics.html#when-do-i-construct-a-session-when-do-i-commit-it-and-when-do-i-close-it

Therefore, I hope you'll find this post before it's too late so you can use SQLAlchemy sessions *the right way*. Cheers 😉


SQLAlchemy: stop calling `session.commit()`


Four years ago, I've made the [first commit](https://github.com/fastapi-users/fastapi-users/commit/4e0b0f6f7d58b55fade54fceea74d2aaa867a17e) to [FastAPI Users](https://github.com/fastapi-users/fastapi-users). At that time, all I wanted was a simple solution to manage users authentication in a FastAPI project. I didn't really have any specific experience in this field, other than what I knew from what other web frameworks like Django did.

Over time, this project grew in popularity and, to my own surprise, I started to actually *like* all the problematics and challenge related to authentication. Quickly, I created other related projects, like [httpx-oauth](https://github.com/frankie567/httpx-oauth) before tackling a very big challenge, [Fief](https://www.fief.dev/).

I don't think it's necessary to explain you why, in this kind of projects, it's critical to [hash users passwords](https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html#background). It's actually fun to see that the first commit of FastAPI Users already has a function to hash passwords using [`passlib`](https://foss.heptapod.net/python-libs/passlib).
This part hasn't changed much since, and you can find a [very similar impIementation](https://github.com/fief-dev/fief/blob/8bd6666f9c492b107f72762e88263d2bb7de2e92/fief/crypto/password.py) in Fief. For years, `passlib` has been the de-facto standard to hash passwords in Python, thanks to its wide algorithms support and well-designed helpers.

However, today, let me introduce you [`pwdlib`](https://github.com/frankie567/pwdlib), a modern password hash helper for Python.

# Why `pwdlib`?

To my great regret, `passlib` has not been very active recently and its [maintenance status is under question](https://foss.heptapod.net/python-libs/passlib/-/issues/187). Starting Python 3.13, unless it's updated, `passlib` won't work anymore.

Currently, it's a [major blocker](https://github.com/fastapi-users/fastapi-users/issues/1325) for FastAPI Users and probably for a lot of other projects too.

That's why I decided to start `pwdlib`, a password hash helper for the modern Python era. However, it's **not designed to be a complete replacement** for `passlib`, which supports numerous [hashing algorithms and features](https://passlib.readthedocs.io/en/stable/lib/index.html).

**✅ Goals**

- [x] Provide an easy-to-use wrapper to hash and verify passwords
- [x] Support modern and secure algorithms like Argon2 or Bcrypt

**❌ Non-goals**

- [ ] Support legacy hashing algorithms like MD5
- [ ] Implement algorithms directly — we should only rely on existing and battle-tested implementations

# Quickstart

The first version I release today supports Argon2 and Bcrypt, and exposes a convenient helper to hash and verify passwords.

```sh
pip install 'pwdlib[argon2]'
```

```py
from pwdlib import PasswordHash

password_hash = PasswordHash.recommended()
hash = password_hash.hash("herminetincture")
password_hash.verify(hash, "herminetincture")  # True
```

If you wish to know more, make sure to [read the documentation](https://frankie567.github.io/pwdlib/).

#	What now?

The first step will be to use it in FastAPI Users and Fief, and adjust the API if needed given the needs and field experience.

I'm also considering to support other common algorithms, like `scrypt`, which is the second choice after Argon2; following [OWASP recommendations](https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html#introduction).

But I also have more global ambitions!

# Building an ecosystem dedicated to users authentication

Currently, projects I maintain are roughly structured like this:

```mermaid
flowchart TD
    PWDLIB[pwdlib]
    HTTPX_OAUTH[HTTPX OAuth]
    FASTAPI_USERS[FastAPI Users]
    FIEF[Fief]
    PWDLIB --> FASTAPI_USERS
    PWDLIB --> FIEF
    HTTPX_OAUTH --> FASTAPI_USERS
    HTTPX_OAUTH --> FIEF
```

There are actually lot of duplicated logic in both FastAPI Users and Fief, and some that would be super-interesting to have access to on their own, without being tied to a framework or software.

My ultimate goal would be to create a `users-core` library (name to be confirmed): a sans-IO project which would expose all the logic and protocols for users authentication: registration, password reset, social connect, passkeys, OpenID Connect, etc. Those implementations could then be used regardless of the framework, whether it's FastAPI, Starlette or Django.

```mermaid
flowchart TD
    style USERS_CORE stroke-width:2px,stroke-dasharray:5 5
    style FRAMEWORK stroke-width:2px,stroke-dasharray:5 5,opacity:20%
    PWDLIB[pwdlib]
    HTTPX_OAUTH[HTTPX OAuth]
    FASTAPI_USERS[FastAPI Users]
    FIEF[Fief]
    USERS_CORE[Users Core]
    FRAMEWORK[...]
    PWDLIB --> USERS_CORE
    HTTPX_OAUTH --> USERS_CORE
    USERS_CORE --> FASTAPI_USERS
    USERS_CORE --> FIEF
    USERS_CORE --> FRAMEWORK
```

I'm very excited about this vision! This is something I have in mind for quite a time now and I really believe it'll help the Python community to build web projects faster, with rock-solid authentication foundations!


Introducing pwdlib, a modern password hash helper for Python


If, like me, you're a huge fan of what [Pydantic](https://docs.pydantic.dev/latest/) and [FastAPI](https://fastapi.tiangolo.com/) folks are doing, you can't have missed the new [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated) construct, which was introduced in Python 3.9. In a few months, we've seen lot of code examples changed from this:

```python
from uuid import uuid4

from pydantic import BaseModel, Field


class User(BaseModel):
    id: str = Field(default_factory=lambda: uuid4().hex)
```

to this:

```python
from uuid import uuid4

from typing_extensions import Annotated

from pydantic import BaseModel, Field


class User(BaseModel):
    id: Annotated[str, Field(default_factory=lambda: uuid4().hex)]
```

*This is taken directly from [Pydantic documentation](https://docs.pydantic.dev/latest/concepts/fields/#using-annotated).*

At first glance, it doesn't seem really useful or practical. But trust me, `Annotated` can grant *super-powers*. Pydantic and friends quickly understood that and they're already rely on it a lot. By the end of this post, you'll probably understand why 😉

> 👋 If you don't really know about type hints in Python, be sure to [check my previous post](/blog/write-better-python-code-faster-with-type-hints) before!

# Motivation 🥱

So, as you know, type hints gives us direct feedback about the validity of our types while coding. It also helps IDE give us smart auto-completions, which is very handy to list methods and their arguments. So, if we consider again the Pydantic's `User` class:

```py
class User(BaseModel):
    id: str = Field(default_factory=lambda: uuid4().hex)

u = User()
print(u.id)
```

`mypy` and friends know that `u.id` exists *and* that it's a string.

Great! But as you noticed, type annotations are not always enough on their own. It's frequent that we need **runtime information** to finely tune the behavior of the library we're using. Here, we used the `Field` class to tell Pydantic that, by default, we want it to generate an UUID4 for this `id`.

Can you see the problem it causes? We tell the type checker `id` is a string, but we actually assign it to a `Field` object 😱 *We're such scoundrels!*

That's because we use this attribute for two different things actually: tell the type it'll have on an **instance** and tell how we want the **class** to behave for this field. To make the type checker happy, Pydantic sets the `Field` return type to `Any`.

That's actually a very common pattern in Python libraries dealing with data modeling.  For example, this is how [SQLAlchemy ORM](https://docs.sqlalchemy.org/en/20/index.html) solves this problem:

```py
from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String, primary_key=True)
```

They define a special type construct, `Mapped` which expects a generic type and plays well with the `mapped_column` return type. It's more correct semantically, but also a bit more involved for the user. Besides, SQLAlchemy team implemented a complete `mypy` plugin to handle this properly.

Another example of this problem is the dependency injection mechanism of FastAPI:

```py
from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons
```

*This is taken directly from [FastAPI documentation](https://fastapi.tiangolo.com/tutorial/dependencies/#__tabbed_1_4).*

The function `common_parameters` is responsible for retrieving pagination parameters in the URL parameters, and return them as a dictionary. To use it in our API endpoint, we use the `Depends` construct of FastAPI, so it knows it'll have to call this function when we receive a request.

Again, you see here the inconsistency between `commons` being a `dict`, while we're assigning it to a `Depends` object.

# `Annotated` to the rescue 🛟

`Annotated` was purposefully implemented to solve those problems: keep consistent and smart type checking *while* being able to add contextual information for the runtime.

Basically, it works like this:

```py
from typing import Annotated, get_type_hints

class User:
    id: Annotated[str, "This is a metadata"]
```

It's a generic type: the first argument should be the **type** you expect for your argument (here, an `int`). Then, you can have **any number of arguments** of whatever nature; here, the string `"This is a metadata"`.

Type checkers, like `mypy`, will only consider the first argument: the type. The rest can be used (or not) by the actual Python program's logic.

How? Well, like any other type annotations, it can be retrieved using the `get_type_hints` function:

```py
type_hints = get_type_hints(func, include_extras=True)
print(type_hints)  # {'id': typing.Annotated[str, 'This is a metadata']}
print(type_hints["id"].__metadata__)  # ('This is a metadata',)
```

Note that we need to set the `include_extras` argument to `True` to get metadata associated with `Annotated`. Then, once we have an `Annotated` object, we can retrieve this metadata as a tuple through the `__metadata__` attribute.

Now, things can go really wild!

# Working example ⚙️

Let's see how we can implement a class to define a data model with default values for attributes, using `Annotated`. The result we want to achieve will look like this:

```py
class User(DefaultModel):
    id: Annotated[str, Default(factory=lambda: uuid4().hex)]
    name: Annotated[str, Default(value="Anonymous")]
```

We'll be able to create data models and declare attributes with their types and default values if they're not provided by the user.

Since `Annotated` metadata can be anything, a clever and convenient way is to define specific classes that'll allow us to attach some data and behavior. Let's first define a `Default` class:

```py
class Default:
    """
    Class defining a behavior for default value,
    either with a static value or a function factory.
    """

    def __init__(
        self, *, value: Any | None = None, factory: Callable[[], Any] | None = None
    ) -> None:
        self.value = value
        self.factory = factory

    def get_default(self) -> Any:
        if self.value is not None:
            return self.value

        if self.factory is not None:
            return self.factory()

        raise RuntimeError()
```

It's a simple class that accepts either a static value or a factory function, allowing us to generate default values dynamically, like UUID or timestamps.

Now, we'll define a base class for our models, `DefaultModel`, that will define the logic to handle those annotations and fill the defaults.

```py
class DefaultModel:
    """
    Class able to parse type annotations using the `Default` class
    and fill the attributes accordingly if not provided.
    """

    def __init__(self, **kwargs) -> None:
        type_hints = get_type_hints(self, include_extras=True)
        # Iterate through each type-hinted attributes
        for attribute, type_hint in type_hints.items():
            try:
                # The value has been provided by the user
                value = kwargs[attribute]
            except KeyError:
                # Otherwise, look for `Annotated` type hint
                if hasattr(type_hint, "__metadata__"):
                    # Look for the `Default` class in metadata
                    for metadata in type_hint.__metadata__:
                        if isinstance(metadata, Default):
                            # Generate the default value
                            value = metadata.get_default()
            finally:
                # Assign the value!
                setattr(self, attribute, value)
```

The logic is quite straightforward: we iterate through each type-hinted attributes and check if the user has provided a value for it in `kwargs`. If not, we'll explore the `Annotated` type hint and look for `Default` classes. It's important to check the nature of the type hint and metadata, because you can't be sure how the user will actually annotate the final class.

Aaaand... That's about it 🙃 We now have reusable logic to assign default values to class attributes. We can check it works:

```py
class User(DefaultModel):
    id: Annotated[str, Default(factory=lambda: uuid4().hex)]
    name: Annotated[str, Default(value="Anonymous")]


u = User(id="1", name="Arthur")
print(u.id)  # 1
print(u.name)  # Arthur

u_default = User()
print(u_default.id)  # 2d10cd0e9a574ecf817ee02e20a0cf45
print(u_default.name)  # Anonymous
```

Neat, isn't it?

Admittedly, unless you're building Python libraries which requires such nice DX, you'll probably rarely have to implement interpretation logic for `Annotated`. But it's more or less how it works under the hood in Pydantic and FastAPI!

# Usage in the wild 🌍

As we said, this pattern has already made its way into major Python libraries.

## In Pydantic

If we take again our Pydantic example, the `Field` construct can now be added as a metadata in `Annotated`:

```py
class User(BaseModel):
    id: Annotated[str, Field(default_factory=lambda: uuid4().hex)]
```

Another advantage of this technique is that we can assign it to a type and reuse it; which was not possible before:

```py
HexUUID = Annotated[str, Field(default_factory=lambda: uuid4().hex)]

class User(BaseModel):
    id: HexUUID

class Post(BaseModel):
    id: HexUUID
```

The type checker still knows it should type it as a string, but from our point-of-view, it's like using a class type!

And it can go even further: Pydantic allows us to [define completely custom types](https://docs.pydantic.dev/latest/concepts/types/#custom-types) with various constraints and behavior:

```py
TruncatedFloat = Annotated[
    float,
    AfterValidator(lambda x: round(x, 1)),
    PlainSerializer(lambda x: f'{x:.1e}', return_type=str),
    WithJsonSchema({'type': 'string'}, mode='serialization'),
]
```

*This is taken directly from [Pydantic documentation](https://docs.pydantic.dev/latest/concepts/types/#adding-validation-and-serialization).*

## In FastAPI

Looking again at the `common_parameters` dependency example, `Annotated` will turn it into this:

```py
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
```

`Depends` is now used as an `Annotated` metadata. Of course, the strength of this approach is that we can assign this to a type and reuse it. Much shorter than before!

```py
CommonParameters = Annotated[dict, Depends(common_parameters)]

@app.get("/items/")
async def read_items(commons: CommonParameters):
    return commons

@app.get("/orders/")
async def read_orders(commons: CommonParameters):
    return commons
```

## In SQLAlchemy

Did you ever repeated a column definition; like a datetime column or a primary key, with all its constraints and stuff, in SQLAlchemy? I'm sure you did! No more with `Annotated`:

```py
intpk = Annotated[int, mapped_column(primary_key=True)]
timestamp = Annotated[
    datetime.datetime,
    mapped_column(nullable=False, server_default=func.CURRENT_TIMESTAMP()),
]
required_name = Annotated[str, mapped_column(String(30), nullable=False)]
```

```py
class Base(DeclarativeBase):
    pass


class SomeClass(Base):
    __tablename__ = "some_table"

    id: Mapped[intpk]
    name: Mapped[required_name]
    created_at: Mapped[timestamp]
```

*This is taken directly from [SQLAlchemy documentation](https://docs.sqlalchemy.org/en/20/orm/declarative_tables.html#mapping-whole-column-declarations-to-python-types).*

Awesome! We have a short, easy-to-use and type-checking compliant way to declare column definitions and reuse it in dozens of models.

---

That's a wrap! I hope this article helped to demystify `Annotated`, which can be surprising at first glance. This new construct opens a whole new world of possibilities in Python, rooted in the type-annotated era. We're only at the beginning: I bet lot of [new libraries will emerge](https://twitter.com/fvoron/status/1751535653274173715) with those ideas in mind.


typing.Annotated, the new Python cool



If you've followed the Python evolutions for the last few years, you probably know that **type hints** are a thing. Since its introduction in Python 3.5, the [`typing` module](https://docs.python.org/3/library/typing.html) has quickly grow and lot of Python codebases nowadays are using type hints.

If you're not using them already, I can guarantee that you will by the end of this post 😉

# The problem 😩

So, as you know, Python is a **dynamically-typed language**. It means that, contrary to compiled languages like C or Rust, variables types are guessed at runtime instead of being validated during a compilation step. Therefore, you can very well write something like this:

```python
a = 42
assert isinstance(a, int)  # `a` is an integer

a = "Polar is so cool"
assert isinstance(a, str)  # `a` is now a string
```

That's convenient because you don't need to explicitly declare the type of the variable. Thank you Python, you're so nice 🥰 Now, let's take a more concrete example:

```python
def sum(a, b):
	return a + b

sum(42, "I'm a string, lol ( ͡° ͜ʖ ͡°)")
```

We define a simple function that should return the sum of two numbers. Then, we call it with an integer and... a string 🤦‍♂️ From Python's point-of-view, this is perfectly valid code. However, it miserably fails at runtime:

```
TypeError: unsupported operand type(s) for +: 'int' and 'str'
```

Of course, we can't add an integer and a string. Obvious, but we had to run it to spot the mistake. Another one? Imagine we have the following class:

```python
class TripToFarFarAway:
    def __init__(self, distance):
        self.distance = distance
        self.progress = 0

    def are_we_there_yet(self):
        return self.progress < self.distance

    def move_forward(self, distance):
        self.progress += distance
```

Let's use it:

```python
trip = TripToFarFarAway(1_000_000)
if not trip.are_we_there():
	print("NOOOO!")
```

Do you see it? No? Python neither... Until we run it:

```
    if not trip.are_we_there():
           ^^^^^^^^^^^^^^^^^
AttributeError: 'TripToFarFarAway' object has no attribute 'are_we_there'. Did you mean: 'are_we_there_yet'?
```

Damned! We misspelled the method's name!

Those are pretty basics examples, but I'm sure you were already confronted to this kind of mistakes; especially in much bigger codebases with dozens of classes, functions, each one with their massive set of arguments. *"Huh, what's the method's name already?"*, *"Oh right, the fourth argument is a string, not an int...*". We all lost *hours* (and our sanity) in those kind of pointless problems. Couldn't we do better? Actually, **we can**.

# Introducing type hints ✍️

The idea is quite simple: add, directly in the code, information about the type we expect on our variables, arguments, properties, etc... A bit similar to what we would do in a compiled language. That's **type hints** (or **type annotations**) .

If we consider our `sum` function again, it would look like this with annotations:

```python
def sum(a: int, b: int) -> int:
	return a + b
```

It looks quite natural: after each argument, we specify the type we expect after a colon `a: int`. The return type is set using an arrow: `-> int`. Cool! So what happens if we run `sum(42, "I'm a string, lol ( ͡° ͜ʖ ͡°)")` again?

```
TypeError: unsupported operand type(s) for +: 'int' and 'str'
```

Oh... It didn't change anything ☹️ That's perfectly normal: Python **is** and **will remain** a **dynamically-typed** language. It means the Python interpreter will simply **ignore the annotations** and do nothing with it. So, how can we use them?

# Introducing static type checkers 🔬

A static type checker is a program able to **parse and analyze your source code** to tell you if your code is consistent regarding typing. For Python, the reference type checker is [`mypy`](https://github.com/python/mypy). Give it a script as input, and it'll check that your type annotations are respected. Let's see this with our `sum` example:

```bash
$ mypy sum.py
sum.py:4: error: Argument 2 to "sum" has incompatible type "str"; expected "int"  [arg-type]
Found 1 error in 1 file (checked 1 source file)
```

Wow, that's interesting! `mypy` tells us that the second argument is a string, while we expect an integer. It's worth to note that `mypy` doesn't even need to actually run your program: it infers everything from the type hints 🧠

With this tool, we have a way to know quickly if we're making a mistake. Of course, it integrates very well with IDE like Visual Studio Code, so the feedback directly shows up while you're coding!

![Typing error in EDE](/posts/images/write-better-python-code-faster-with-type-hints/typing_error.png)

Talking about IDE, some also ship with their own static type checkers so they can offer smart autocompletions or imports. Visual Studio Code (again), uses Pyright to show you things like this:

![Autocompletion in IDE](/posts/images/write-better-python-code-faster-with-type-hints/autocompletion.png)

See? VS Code automatically suggests methods and properties from the `TripToFarFarAway` class. No more back and forth with the class definition or the library' docs!

Needless to say those tools will help you to:

* Save time
* *Dramatically* reduce errors

# More complex type hints 📈

The example we've seen so far was pretty simple, but in the real world, we use much more than just scalar types like `int` or `str`. What about data structures? Callables? Generators? Don't worry, they are supported to!

Let's see how we can define a **list of integers** and a **dictionary mapping strings to integers**.

```python
l: list[int] = [1, 2, 3, 4, 5]
d: dict[str, int] = {"a": 1, "b": 2, "c": 3}
```

Again, we define the type after a colon. Here we use the builtin types `list` and `dict` to which we add a type variable between square brackets. That's quite close to the concept of [Generics](https://en.wikipedia.org/wiki/Generic_programming). Now, type checkers will now that `l` contains only integers and `d` should only map strings to integers. So the following code:

```python
# Try to append a string to `l`
l.append("I'm a string, lol ( ͡° ͜ʖ ͡°)")
# Try to sum an item of `d` with a string
d["a"] + "I'm a string, lol ( ͡° ͜ʖ ͡°)"
```

...will produce the following errors with `mypy`:

```
type_hints.py:5: error: Argument 1 to "append" of "list" has incompatible type "str"; expected "int"  [arg-type]
type_hints.py:7: error: Unsupported operand types for + ("int" and "str")  [operator]
```

As we said, Python will remain a dynamically-typed language. So we can very well have a list with elements of different types, like a list with both `int` and `float`. This can be expressed like this in annotations:

```py
l: list[int | float] = [42, 3.14, 1337]
```

With the pipe `|`, we specify an **union** of possible types. That one is also very useful for **optional arguments** in functions:

```py
def greeting(name: str | None = None) -> str:
        return f"Hello, {name if name else 'Anonymous'}"
```

Note how the type annotation for `None` is just... `None`!

> **Type annotations were different before Python 3.9**
>
> Before Python 3.9, it wasn’t possible to annotate lists, tuples, sets, and dictionaries using the builtin type. Special construct from `typing` module were needed: `l: typing.List[int] = [1, 2, 3, 4, 5]`.
>
> Likewise, the `|` wasn't available: `typing.Union[int, float]`.
>
> This way of annotating is now deprecated, but you may still find it in backwards-compatible code.

Of course, we can go way further than that, but we'll stop here for this introduction. If you want to go further, `mypy` documentation is a must-read: [https://mypy.readthedocs.io/en/stable/index.html](https://mypy.readthedocs.io/en/stable/index.html)

# Introspection 😌

In programming, introspection is the ability for a program to get information about an object **at runtime**. As you may know, Python is *very* good at this. It even has a [whole module in the standard library](https://docs.python.org/3/library/inspect.html) dedicated to this. A small example:

```py
import inspect

def sum(a, b):
    """Returns the sum of two integers."""
    return a + b

sum_signature = inspect.signature(sum)
print(sum_signature.parameters)
print(inspect.getdoc(sum))
```

The output of this program is the following:

```
OrderedDict({'a': <Parameter "a">, 'b': <Parameter "b">})
Returns the sum of two integers.
```

Impressive! Python is able to give us the signature of the function, and even the attached docstring! Do you guess why I'm talking about this? Because we can **introspect type hints** too 🤯

Let's try again with the type annotated version of `sum`:

```py
def sum(a: int, b: int) -> int:
    """
    Returns the sum of two integers.
    """
    return a + b

sum_signature = inspect.signature(sum)
print(sum_signature.parameters)
```

```
OrderedDict({'a': <Parameter "a: int">, 'b': <Parameter "b: int">})
```

The signature is able to give us an ordered dictionary of all the arguments expected by the function. Each one is a [`Parameter`](https://docs.python.org/3/library/inspect.html#inspect.Parameter) object and **it does retain the type annotation**:

```py
print(sum_signature.parameters["a"].annotation)
```

```
<class 'int'>
```

Boom! `a` should be an `int`. That's a **game changer**: it means that we can actually use this information at runtime to change the behavior of our program.

A quite natural example could be to implement a decorator to checks the validity of a function arguments:

```py
import functools
import inspect

def check_arguments_type(f):
    parameters = inspect.signature(f).parameters
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        # Check args
        for arg, (parameter_name, parameter) in zip(args, parameters.items()):
            assert isinstance(arg, parameter.annotation), f"{parameter_name} should be {parameter.annotation}"

        # Check kwargs
        for key, value in kwargs.items():
            parameter = parameters[key]
            assert isinstance(value, parameter.annotation), f"{key} should be {parameter.annotation}"

        return f(*args, **kwargs)
    return wrapper

@check_arguments_type
def sum(a: int, b: int) -> int:
    """
    Returns the sum of two integers.
    """
    return a + b

sum(42, "I'm a string, lol ( ͡° ͜ʖ ͡°)")
```

> ⚠️ This is a very naive approach with lot of edge cases — don't use it like this in your projects!

Now, we have a type check at runtime!

```
Traceback (most recent call last):
  File "sum.py", line 29, in <module>
    sum(42, "I'm a string, lol ( ͡° ͜ʖ ͡°)")
  File "sum.py", line 11, in wrapper
    assert isinstance(arg, parameter.annotation), f"{parameter_name} should be {parameter.annotation}"
AssertionError: b should be <class 'int'>
```

This example was focused on functions, but it works with any Python objects accepting type hints. In general, we can use the function [`typing.get_type_hints()`](https://docs.python.org/3/library/typing.html#typing.get_type_hints) to get annotations for any arbitrary object:

```py
class Movie:
    title: str
    duration_minutes: int

print(typing.get_type_hints(Movie))
```

```
{'title': <class 'str'>, 'duration_minutes': <class 'int'>}
```

## How it's used in the real-world?

This technique is really clever and powerful, and it actually made its way into major Python libraries and frameworks. In particular, this is at the heart of [Pydantic](https://docs.pydantic.dev/latest/), the data validation library. With it, you define models using type annotations:

```py
class Delivery(BaseModel):
    timestamp: datetime
    dimensions: tuple[int, int]
```

And it'll use them at runtime to perform complex data validation and conversion!

Another great example is [FastAPI](https://fastapi.tiangolo.com/), which automatically determines the input data expected by an API endpoint using type hints (and Pydantic models 😉):

```py
@app.post("/items/")
async def create_item(item: Item):
    return item
```

In this example, FastAPI will:

* Generate an API schema
* Validate the JSON input so it conforms to the `Item` model
* Give you a ready-to-use instance of `Item`.

The cherry-on-the-cake of those approaches is that it gives us good type checking out of the box!

---

That's all I wanted to share with you about Python type hints! If you didn't know it, I hope it'll give you good reasons to start using them.

In my next post, we'll focus on a recent and *really powerful* typing construct: `Annotated`! See you soon 👋


Write better Python code faster with type hints


[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