Forem: Timothée Mazzucotelli

Testing a FastAPI application using Ormar models and Alembic migrations

Timothée Mazzucotelli — Sat, 22 Oct 2022 18:26:59 +0000

In the previous post
I showed how to add Alembic migrations to an existing FastAPI + Ormar project.
In this post we will see how to write unit tests for such applications.

We start with the following project layout:

./
    src/
        project/
            __init__.py
            app.py
            models.py
    tests/

Database models

Let say we have three models: Artist, Album and Track.
To keep things simple, we just add a name field on each.

"""Database models."""

import os
import databases
import ormar
import sqlalchemy

SQLITE_DB = os.getenv("SQLITE_DB", "sqlite:///db.sqlite")
DB_PATH = SQLITE_DB.replace("sqlite:///", "")


class BaseMeta(ormar.ModelMeta):
    database = databases.Database(SQLITE_DB)
    metadata = sqlalchemy.MetaData()


class Artist(ormar.Model):
    class Meta(BaseMeta): ...

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)


class Album(ormar.Model):
    class Meta(BaseMeta): ...

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    artist: Artist = ormar.ForeignKey(Artist, nullable=False)


class Track(ormar.Model):
    class Meta(BaseMeta): ...

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=100)
    album: Album = ormar.ForeignKey(Album, nullable=False)

Database and migrations helpers

Now lets create helpers to easily (re)create, update (migrate) or stamp the database.
We will put everything that is related to migrations in a migrations subpackage:

./
    src/
        project/
            migrations/
                __init__.py
            __init__.py
            app.py
            models.py
    tests/

We will define helpers in the __init__ module.
Loguru will be used to log things,
but that's optional and you can remove logging lines or use another logging framework.

"""Database migrations modules."""

from functools import wraps
from pathlib import Path

import sqlalchemy
from alembic import command as alembic
from alembic.config import Config
from loguru import logger

from project.models import DB_PATH, SQLITE_DB, BaseMeta


def get_alembic_config(db_url: str = SQLITE_DB) -> Config:
    alembic_cfg = Config()
    alembic_cfg.set_main_option("script_location", "project:migrations")
    alembic_cfg.set_main_option("sqlalchemy.url", str(db_url))
    return alembic_cfg


def upgrade_database(revision: str = "head", db_url: str = SQLITE_DB) -> None:
    alembic_cfg = get_alembic_config(db_url)
    alembic.upgrade(alembic_cfg, revision)


def stamp_database(revision: str = "head", db_url: str = SQLITE_DB) -> None:
    alembic_cfg = get_alembic_config(db_url)
    alembic.stamp(alembic_cfg, revision)


def create_database(db_url: str = SQLITE_DB) -> None:
    engine = sqlalchemy.create_engine(db_url, connect_args={"timeout": 30})
    BaseMeta.metadata.create_all(engine)


def db_lock(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        lock = Path(DB_PATH).parent / ".dblock"
        try:
            lock.mkdir(parents=True, exist_ok=False)
        except FileExistsError:
            logger.debug("Migrations are already being applied")
            return
        logger.debug("Applying migrations")
        try:
            func(*args, **kwargs)
        finally:
            lock.rmdir()

    return wrapper


@db_lock
def apply_migrations(db_url: str = SQLITE_DB) -> None:
    if Path(DB_PATH).exists():
        upgrade_database(db_url=db_url)
    else:
        create_database(db_url=db_url)
        stamp_database(db_url=db_url)

Note how each function accepts a db_url parameter:
this will be very useful to support different environments,
such as development, production and testing.

We still need the Alembic configuration module:

./
    src/
        project/
            migrations/
                __init__.py
                env.py
            __init__.py
            app.py
            models.py
    tests/

import os

from alembic import context
from sqlalchemy import engine_from_config, pool

from project.models import BaseMeta

config = context.config
target_metadata = BaseMeta.metadata


def get_url():
    # allow configuring the database URL / filepath using an env var, useful for production
    return os.getenv("SQLITE_DB", config.get_main_option("sqlalchemy.url"))


def run_migrations_offline():
    url = get_url()
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        render_as_batch=True,  # needed for sqlite backend
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online():
    configuration = config.get_section(config.config_ini_section)
    configuration["sqlalchemy.url"] = get_url()
    connectable = engine_from_config(
        configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            render_as_batch=True,  # needed for sqlite backend
        )
        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Automatic database creation/update

Now, lets configure our FastAPI app so that the database is automatically
created or updated every time we run our app (using uvicorn for example):

./
    src/
        project/
            migrations/
                __init__.py
                env.py
            __init__.py
            app.py
            models.py
    tests/

"""FastAPI application."""

from fastapi import FastAPI

from project.migrations import apply_migrations
from project.models import BaseMeta

app = FastAPI()


@app.on_event("startup")
async def startup() -> None:
    apply_migrations(str(BaseMeta.database.url))
    if not BaseMeta.database.is_connected:
        await BaseMeta.database.connect()


@app.on_event("shutdown")
async def shutdown() -> None:
    if BaseMeta.database.is_connected:
        await BaseMeta.database.disconnect()

Creating or updating the database in the startup event allows several things:

in a development environment, developers can simply run the server, and the database is automatically created. They don't have to worry about running a database creation command. Similarly, they can simply delete the db.sqlite file and restart the server to empty their local database, or copy a pre-populated SQlite file to reset their local database to a particular state.
in a production environment, in which you have no control (no possibility to run custom shell commands), migrations are applied automatically upon starting the server. If there are no (new) migrations, the startup event is a no-op: nothing happens and the server starts normally, with the current database untouched. Note that we ensure only one instance of the server will apply the migrations, as to prevent multiple parallel/concurrent accesses to a potentially shared storage space (for example multiple pods accessing the same persistent volume on a Kubernetes infrastructure).
in a testing environment, tests will be able to provide a unique database URL (a local file path) so that they each have their own temporary database. It means tests will be able to run in parallel, for example using pytest-xdist.

Pytest fixture

Now lets create a Pytest fixture that will allow each test
to get access to its own unique, temporary database:

./
    src/
        project/
            migrations/
                __init__.py
                env.py
            __init__.py
            app.py
            models.py
    tests/
        __init__.py
        conftest.py

"""Configuration for the pytest test suite."""

import os
from pathlib import Path

import databases
import pytest
from asgi_lifespan import LifespanManager
from httpx import AsyncClient

from project.app import app
from project.migrations import create_database, stamp_database
from project.models import BaseMeta


@pytest.fixture()
async def async_client(tmp_path: Path, monkeypatch):
    """
    Provide an HTTPX asynchronous HTTP client, bound to an app using a unique, temporary database.

    Arguments:
        tmp_path: Pytest fixture: points to a temporary directory.
        monkeypatch: Pytest fixture: allows to monkeypatch objects.

    Yields:
        An instance of AsyncClient using the FastAPI ASGI application.
    """
    db_url = f"sqlite:///{tmp_path}/db.sqlite"
    create_database(db_url=db_url)
    stamp_database(db_url=db_url)
    database = databases.Database(db_url)
    monkeypatch.setattr(BaseMeta, "database", database)

    lifespan = LifespanManager(app)
    httpx_client = AsyncClient(app=app, base_url="http://testserver")

    async with httpx_client as client, lifespan:
        yield client

You'll notice that we use asgi-lifespan.
Without it, the startup and shutdown ASGI events would not be triggered.

In the startup event, we apply migrations using the URL in BaseMeta.database.url.
This allows us to monkeypatch the database attribute in our fixture to change
the database URL for each test.

Model instances factories

In our tests, we'll want to insert some rows in the database to test our API.
Doing so manually can be cumbersome, as you have to define each instance
one after the other, linking them together.
To ease the process, we use factory-boy,
with which we'll be able to define model factories. With these factories,
it will be very easy to create instances of models in our tests.

./
    src/
        project/
            migrations/
                __init__.py
                env.py
            __init__.py
            app.py
            models.py
    tests/
        __init__.py
        conftest.py
        factories.py

"""Factory classes to build models instances easily."""

import factory

from project import models


class ArtistFactory(factory.Factory):
    class Meta:
        model = models.Artist

    id = 1
    name = "artist name"


class AlbumFactory(factory.Factory):
    class Meta:
        model = models.Album

    id = 1
    name = "album name"
    artist = factory.SubFactory(ArtistFactory)


class TrackFactory(factory.Factory):
    class Meta:
        model = models.Track

    id = 1
    name = "track name"
    album = factory.SubFactory(AlbumFactory)

With these factories you can now create an artist, album and track,
all linked together, using a single line of code:

from tests import factories

track = factories.TrackFactory()

You can change arbitrary attributes when creating instances:

track = factories.TrackFactory(
    name="other track name",
    album__name="other album name",
    album__artist__name="other artist name",
)

Refer to factory-boy's documentation for more examples.
You could also use Faker
to set more relevant default values to your instances attributes.

Populating the database with data

Creating instances is nice, but they are not magically inserted in the database for us.
Since instances are Ormar model instances, we could technically use the save() method
on the instances we create to save them in the database, however I did not try that
and cannot guarantee it will work for multiple instances linked together at once.

Instead, and only if you have added
CRUD operations
to your API, you can call your API routes to create the instances in the database.

For this, I chose to create a new helper module,
but that's probably not the best design you can come up with,
so feel free to discard the next suggestions and follow your instincts.

./
    src/
        project/
            migrations/
                __init__.py
                env.py
            __init__.py
            app.py
            models.py
    tests/
        __init__.py
        conftest.py
        factories.py
        helpers.py

"""Helpers for tests."""

from project.models import Artist, Album, Track
from tests import factories


async def create_artist(client) -> Artist:
    artist = factories.ArtistFactory()

    payload = artist.dict()
    response = await client.post("/artists", json=payload)
    response.raise_for_status()

    return artist


async def create_album(client) -> Album:
    album = factories.AlbumFactory()

    # create artist first
    payload = album.artist.dict()
    response = await client.post("/artists", json=payload)
    response.raise_for_status()

    # then create album
    payload = album.dict()
    response = await client.post("/albums", json=payload)
    response.raise_for_status()

    return album


async def create_track(client) -> Track:
    track = factories.TrackFactory()

    # create artist first
    payload = track.album.artist.dict()
    response = await client.post("/artists", json=payload)
    response.raise_for_status()

    # then create album
    payload = track.album.dict()
    response = await client.post("/albums", json=payload)
    response.raise_for_status()

    # finally create track
    payload = track.dict()
    response = await client.post("/tracks", json=payload)
    response.raise_for_status()

    return track

Example tests

Now you can easily populate the database in tests,
and call other API routes to test their behavior and output.

./
    src/
        project/
            migrations/
                __init__.py
                env.py
            __init__.py
            app.py
            models.py
    tests/
        __init__.py
        conftest.py
        factories.py
        helpers.py
        test_tracks.py

"""Tests for the `tracks` routes."""

import pytest

from tests import factories, helpers


@pytest.mark.asyncio()
async def test_tracks_create(async_client):
    track = await helpers.create_track(async_client)
    # ...then test other API routes

Note how we use the previously defined fixture async_client.
Just adding that fixture as a parameter to our test function
ensures we have a temporary, dedicated database for this test.

How to add Alembic migrations to an existing FastAPI + Ormar project

Timothée Mazzucotelli — Sat, 22 Oct 2022 18:23:08 +0000

I had trouble adding an ON DELETE CASCADE constraint to foreign keys after they were already created, so here's a small post explaning how I did it. This specific constraint, ON DELETE CASCADE, is not relevant here. This post will help you for any kind of change you want to apply on database already used in production.

First, I knew I was going to need "migrations", because my database was already used in production. I didn't previously add any migration. so I had to create the initial one, reflecting the current state of the database, and a second one to add the cascade delete constraints.

As documented in Ormar's pages themselves, the tool to write migrations is Alembic. Alembic has an "autogenerate" feature that can compare the "metadata" of the database (an SQLAlchemy object in memory) against the actual database to generate a migration file.

I'm using PDM to manage my dependencies, so I added Alembic to them with:

pdm add alembic

I did not add it to development dependencies because I can't SSH into the production machine to run a one-off command. It has to be a production dependency so I can run migrations (upgrade the state of the database) every time the app is deployed and started.

Next, following the Alembic tutorial, I initialized it with:

pdm run alembic init alembic

I then had to modify a bit the alembic.ini and alembic/env.py files it created, to set the URL of the database and to import the right metadata object:

# works locally only, see later how to deal with multiple environments
sqlalchemy.url = sqlite:///db.sqlite

from my_app.db import metadata
target_metadata = metadata

I was now able to generate the initial migration with the following command, making sure I removed the ondelete="CASCADE" arguments from my models' foreign keys:

pdm run alembic revision --autogenerate -m "Initial migration."

The first issue I had was that Alembic was generating a migration that would drop tables instead of creating them. After some investigation, I realized that it was because the metadata object I imported was empty. With Ormar, the metadata object is populated only when loading the models definitions. So instead of importing it from my db module which only instantiated it, I imported it from my models module which uses it to declare my models classes:

from my_app.models import metadata
target_metadata = metadata

I deleted the broken migration, and recreated a correct one:

pdm run alembic revision --autogenerate -m "Initial migration."

To mark this migration as "applied", I ran the Alembic upgrade command:

pdm run alembic upgrade head

Now it was time to add the second migration, the one that adds the "on delete cascade" constraint. I added the ondelete="CASCADE" arguments to my models' foreign keys, add created the second migration:

pdm run alembic revision --autogenerate -m 'Add `ondelete="CASCADE"` to foreign keys.'

But then I got this error:

NotImplementedError: No support for ALTER of constraints in SQLite dialect.
Please refer to the batch mode feature which allows for SQLite migrations using a copy-and-move strategy.

Oh no. Thankfully, a quick search on the internet got me to this SO post and answer: you have to use render_as_batch=True to be able to alter constraint when using an SQLite database. In alembic/env.py:

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    render_as_batch=True
)

# in both run_migrations_offline() and run_migrations_online()!

I ran the previous command again, and it worked. I applied the migration with pdm run alembic upgrade head again.

At this point, I have migrations that I will be able to apply on the production database. But how do I do that?

First, I need to support local and production environments. The way to do it is with an environment variable. The SQLite database URL will be stored in the SQLITE_DB environment variable. All I have to do now is to modify alembic/env.py to use this variable first, and fallback to the value configured in alembic.ini:

def get_url():
    return os.getenv("SQLITE_DB", config.get_main_option("sqlalchemy.url"))

# in run_migrations_offline:
    url = get_url()

# in run_migrations_online:
    configuration = config.get_section(config.config_ini_section)
    configuration["sqlalchemy.url"] = get_url()
    connectable = engine_from_config(
        configuration, prefix="sqlalchemy.", poolclass=pool.NullPool,
    )

I was inspired by this FastAPI example.

Now I need to set this variable for production, for example in an .env file. For me it was in a Helm chart:

...
env:
- name: SQLITE_DB
  value: "sqlite:///different/path/to/db.sqlite"

Finally, I have to call the alembic upgrade head command each time I start the application. This can be done in FastAPI's "startup" event:

import os
from pathlib import Path

import sqlalchemy
from alembic.config import Config
from alembic.commad import upgrade
from fastapi import FastAPI

from my_app.db import metadata, database

app = FastAPI()
app.state.database = database

@app.on_event("startup")
async def startup() -> None:
    """Startup application."""
    db_url = os.getenv(SQLITE_DB)
    db_path = db_url.replace("sqlite://", "")
    if not Path(db_path).exists():
        engine = sqlalchemy.create_engine(db_url)
        metadata.drop_all(engine)
        metadata.create_all(engine)

    upgrade(Config(), "head")

    database = app.state.database
    if not database.is_connected:
        await database.connect()

In my next post,
you will see how to write tests for such setups,
but also how to configure all this in a more robust and elegant way.

Local HTTP server sending fake files for testing purposes

Timothée Mazzucotelli — Thu, 17 Dec 2020 18:19:21 +0000

I have developed a Python client/library
for aria2
called aria2p.

To test my code, I spawn aria2c processes
to interact with them and see if everything works correctly.
I know I could mock the calls, but I feel like actually
testing the interaction between my client and aria2c
is a better way to ensure my code does its best.

Until recently the test suite fed actual downloads to aria2c
(like Linux distributions ISOs, metalinks, magnets, or torrents),
but of course the inevitable happened: some URLs broke.

So I decided to go full local.
This post describes my attempt at spawning local HTTP servers
to serve fake/virtual files efficiently,
files that are then fed to aria2c.

The basics: `http.server`

As you may already know, Python's standard library provides
a ready-to-go HTTP server that can serve files from a folder.
It lives in the http.server module, and you can run it in
a single command:

python -m http.server

This will serve the contents of your current working directory
on http://0.0.0.0:8000.

I was already using it in some tests when I needed to feed
very small files to aria2c. I was serving aria2p's code folder
to use the LICENSE file: 759 bytes.

So, the simplest solution that immediately came to mind was:
dump text into files with predefined sizes,
like 1B, 100B, 1KiB, 100KiB, 1MiB, 10MiB, 100MiB, etc.
And do this each time you run the test suite,
in a temporary directory that you will serve using
Python's built-in HTTP server.

But that sounds a bit wasteful, right? Wasting disk cycles on
expensive SSDs is not really attractive.
So it led me to the next solution.

Serving files from memory

Instead of creating files on the disk,
why not just create them in memory?

So I looked at the sources in http.server,
particularly how SimpleHTTPRequestHandler
sends the files contents to the client,
and eventually ended up with this code:

# http_server.py
import shutil
import socketserver
import sys
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler
from io import BytesIO


class RequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        size = self.path.lstrip("/")
        self.send_response(HTTPStatus.OK)
        self.send_header("Content-type", "application/octet-stream")
        self.send_header("Content-Length", size)
        self.end_headers()
        shutil.copyfileobj(BytesIO(b"1" * int(size)), self.wfile)


def run_http_server(port):
    with socketserver.TCPServer(("", port), RequestHandler) as httpd:
        try:
            httpd.serve_forever()
        finally:
            httpd.server_close()


if __name__ == "__main__":
    print("Running HTTP server on port 8000")
    run_http_server(8000)

This snippet can be run directly with python http_server.py.
The run_http_server can also be imported somewhere else.

This server accepts GET requests on the path /{size},
which allows us to choose the size of the file we want to download.
For example: GET http://localhost:8000/1024 for 1Kib.
The files are then only composed of ones.

There is a problem however:
this server will not support a heavy load
(lots of tests running in parallel),
because it can handle only one request at a time.

Multi-threading

Thankfully, socketserver also provides a ThreadingTCPServer
we can use. It's as simple as:

def run_http_server(port=PORT):
    with socketserver.ThreadingTCPServer(("", port), RequestHandler) as httpd:
        try:
            httpd.serve_forever()
        finally:
            httpd.server_close()

Reusable address

Even with the finally clause,
I noticed that the address kept being "in use"
for up to a minute sometimes, after the server was killed.

This can be alleviated with the allow_reuse_address attribute
of the socket server:

class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True


def run_http_server(port=PORT):
    with Server(("", port), RequestHandler) as httpd:
        try:
            httpd.serve_forever()
        finally:
            httpd.server_close()

Still not enough

But even with all this, many tests were still failing,
and the standard error captured by pytest showed
a lot of BrokenPipe errors coming from the server.

I think it's because this server is way too simple,
and handles only GET requests. Maybe aria2c is using
HEAD as well? What about its "pausing" feature,
allowing to pause downloads and resume them?
Will the server handle it correctly?

So, instead of digging more into this,
I decided to try running a proper server.

With FastAPI

I've been using FastAPI recently,
so I gave it a try. A very small amount of code was needed:

# http_server.py
from io import BytesIO

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get("/{size}")
async def get(size):
    return StreamingResponse(BytesIO(b"1" * size), media_type="application/octet-stream")

I went with a StreamingResponse
because the FileResponse only accepts paths to actual files.

To run the server, I run this command: uvicorn http_server:app.
And it seems to work great!
All the tests that were previously failing now pass.

Trying to be clever

I'm still not really satisfied with this solution.
It feels wasteful to allocate memory for each handled request.
I tried storing the BytesIO object into a dictionary,
with sizes as keys, but I guess their buffer get consumed
when sending the response, so this doesn't work
on subsequent requests using the same size.
Instead of storing the BytesIO objects,
we can store the strings themselves!

from io import BytesIO

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()
allocated = {}

@app.get("/{size}")
async def get(size: int):
    if size not in allocated:
        allocated[size] = b"1" * size
    return StreamingResponse(BytesIO(allocated[size]), media_type="application/octet-stream")

But even then, why allocating memory at all?
Couldn't we have a really "virtual" file,
some kind of generator that generates chunks of data,
sent as a streaming response, and that stops at the desired size?

Well the StreamingResponse
accepts async generators,
so here's what I tried:

async def virtual_file(size, chunks=4096):
    while size > 0:
        yield b"1" * min(size, chunks)
        size -= chunks


@app.get("/{size}")
async def get(size: int):
    return StreamingResponse(virtual_file(size), media_type="application/octet-stream")

It seems to work, and the memory footprint is much, much lower,
but I get a lot of socket.send() raised exception. messages from FastAPI,
and a test occasionally fails. Anyway, I think it's good enough.

Additional niceties

To make it easier to choose a medium-to-big file size,
we can accept a unit suffix:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


def translate_size(size):
    try:
        return int(size)
    except ValueError:
        pass
    size = size.lower()
    if size.endswith("k"):
        multiplier = 2 ** 10
    elif size.endswith("m"):
        multiplier = 2 ** 20
    elif size.endswith("g"):
        multiplier = 2 ** 30
    else:
        raise ValueError("size unit not supported:", size)
    return int(size.rstrip("kmg")) * multiplier


async def virtual_file(size, chunks=4096):
    while size > 0:
        yield b"1" * min(size, chunks)
        size -= chunks


@app.get("/{size}")
async def get(size: str):
    return StreamingResponse(
        virtual_file(translate_size(size)),
        media_type="application/octet-stream",
    )

Note that the route size argument is not an integer anymore but a string.
It allows us to send GET requests like /4k or /100m,
values that are then translated to bytes.

Pytest fixture

Since I run tests in parallel thanks to the pytest-xdist plugin,
I must make sure I run only one instance of the HTTP server.

We can accomplish this by checking the worker ID
(to handle both parallel and non-parallel cases)
and by using a lock.

The worker ID will be master for non-parallel runs.
It can be obtained with the worker_id fixture.

For the lock, I chose to use mkdir as it's an atomic operation.
If the directory already exists, the operation fails
(someone else already got the lock).

Once the server is running, we must make sure it's ready,
so we send GET requests until it responds.

We put all this in a function and declare it a "fixture"
with "session" scope and "automatic use":

# tests/conftest.py
import subprocess
import sys
import time

import pytest
import requests  # or httpx


def spawn_and_wait_server(port=8000):
    process = subprocess.Popen([
        sys.executable,
        "-m",
        "uvicorn",
        "tests.http_server:app",
        "--port",
        str(port),
    ])
    while True:
        try:
            requests.get(f"http://localhost:{port}/1")
        except:
            time.sleep(0.1)
        else:
            break
    return process


# credits to pytest-xdist's README
@pytest.fixture(scope="session", autouse=True)
def http_server(tmp_path_factory, worker_id):
    if worker_id == "master":
        # single worker: just run the HTTP server
        process = spawn_and_wait_server()
        yield process
        process.kill()
        process.wait()
        return

    # get the temp directory shared by all workers
    root_tmp_dir = tmp_path_factory.getbasetemp().parent

    # try to get a lock
    lock = root_tmp_dir / "lock"
    try:
        lock.mkdir(exist_ok=False)
    except FileExistsError:
        yield  # failed, don't run the HTTP server
        return

    # got the lock, run the HTTP server
    process = spawn_and_wait_server()
    yield process
    process.kill()
    process.wait()

With this, each time you run your test suite,
exactly one instance of the HTTP server will run.

Passing Makefile arguments to a command, as they are typed in the command line

Timothée Mazzucotelli — Thu, 17 Dec 2020 18:15:12 +0000

In my Python projects,
I use a combination of Duty
(my task runner) and make.

My Makefile declares the same tasks as the ones written in duties.py,
but in a generic manner:

TASKS = check test release

.PHONY: $(TASKS)
$(TASKS):
    @poetry run duty $@

So, instead of running poetry run duty check, I can run make check.
Convenient, right?

Except that some duties (tasks) accept arguments. For example:

poetry run duty release version=0.1.2

So how do I allow the make equivalent?

make release version=0.1.2

Experiments

Do I write a specific rule in the Makefile for the release task?

TASKS = check test release

.PHONY: release
release:
    @poetry run duty release version=$(version)

.PHONY: $(TASKS)
$(TASKS):
    @poetry run duty $@

Meh. My Makefile rules are not generic anymore.
Besides, what if the argument is optional?
If I don't pass it when running make,
the command will end up being poetry run duty release version=,
which is just wrong.

Instead, I'd like to find a generic way to insert the arguments
in the command just as they are typed on the command line:

make release
# => poetry run duty release

make release version=0.1.2
# => poetry run duty release version=0.1.2

Well, after a few hours playing with Makefiles features,
I got a nice solution!

Sorcery

Let me sprinkle this dark magic right here:

args = $(foreach a,$($(subst -,_,$1)_args),$(if $(value $a),$a="$($a)"))

check_args = files
docs_serve_args = host port
release_args = version
test_args = match

TASKS = \
    check \
    docs-serve \
    release \
    test

.PHONY: $(TASKS)
$(TASKS):
    @poetry run duty $@ $(call args,$@)

What happens here?!

Recipe

args = $(foreach a,$($(subst -,_,$1)_args),$(if $(value $a),$a="$($a)"))

The heart of the magic. We declare a function called args.
We later call it with $(call args,$@).

It could be described like this:

args reference := first parameter
replace - by _ in args reference
append "_args" to args reference
get argument names by dereferencing args reference
for each argument name
  get argument value by dereferencing argument name
  if argument value is not empty
    print "argument name = argument value"

This is why we declare our arguments like this:

check_args = files
docs_serve_args = host port
release_args = version
test_args = match

When running make docs-serve host=0.0.0.0,
the args function will do the following:

args_ref := "docs-serve"
args_ref becomes "docs_serve" (replace)
args_ref becomes "docs_serve_args"
args_names is value of "docs_serve_args" variable
args_names therefore is "host port"
arg "host":
  variable "host" is not empty
  print "host=0.0.0.0"
arg "port":
  variable "port" is empty
  print nothing

So when calling $(call args,$@), $@ is replaced
by the rule name, which is docs-serve in this example,
and host=0.0.0.0 is added to the command.

We successfully re-built the arguments passed on the command line!

Side-effects

Arguments can be passed to make in no particular order.
The following commands are all equivalent:

make hello world=earth foo bar=baz
make hello foo bar=baz world=earth
make bar=baz hello foo world=earth

It can be seen as an advantage but as an inconvenient as well,
because you cannot have arguments with the same name for different commands.
Or at least, you could not use these commands and arguments at the same time.

args = $(foreach a,$($(subst -,_,$1)_args),$(if $(value $a),$a="$($a)"))

rule1_args = version
rule2_args = version name

TASKS = rule1 rule2

.PHONY: $(TASKS)
$(TASKS):
    @poetry run duty $@ $(call args,$@)

Here rule1 and rule2 both accept a version argument.

make rule1 version=1  # OK

make rule1 version=1 rule2  # not OK
# it will result in "poetry run duty rule1 version=1"
# then "poetry run duty rule2 version=1"!

make rule1 version=1 rule2 version=2  # ???
# I couldn't get the courage to try more of these dark arts
# so I don't know what would happen here...

Addendum

Not a real addendum.

Share your tricks!

Python packages and plugins as namespace packages

Timothée Mazzucotelli — Thu, 17 Dec 2020 18:12:37 +0000

A user of mkdocstrings
wrote a Crystal handler for their own use-case.
They asked on the Gitter channel
if we could allow to load external handlers, so they don't have to fork
the project and install the fork, but rather just install their
lightweight package containing just the handler.

We both saw Python namespace packages as good candidates,
so I experimented a bit, and here are my conclusions.

Based on the documentation, there are
3 ways of creating a Python namespace package:

native namespace packages
pkgutil-style namespace packages
pkg-resources-style namespace packages

I only considered the first two options (I'll let you read the docs to see why).

Native namespace packages

Native namespace packages are Python 3 compatible only.
Good, I don't support Python 2 in my projects.

To write a native namespace package, you just have to do one thing:
drop the __init__.py!

my_namespace_package/
    my_module_a.py

my_namespace_package/
    my_module_b.py

Install both packages and you will be able to import
both my_module_a and my_module_b from my_namespace_package.

If you want to provide deeper modules, you will have to drop
all __init__.py files along the way:

# main package
package/
    level1/
        level2/
            builtin_module.py
    subpackage/
        __init__.py
        sub.py
    a.py

# plugin
package/
    level1/
        level2/
            plugin_module.py

Notice how we don't have any __init__.py in package, level1 and level2.
This is required for the plugin package to add modules there.

In this example, subpackage has an __init__.py file,
it means that our plugin package could not add a module into it.

`pkgutil`-style namespace packages

The pkgutil way is compatible with Python 2 and Python 3.
It allows to keep __init__.py modules.

# main package
package/
    __init__.py
    level1/
        __init__.py
        level2/
            __init__.py
            builtin_module.py

# plugin
package/
    __init__.py
    level1/
        __init__.py
        level2/
            __init__.py
            plugin_module.py

Each __init__.py must contain this code:

from pkgutil import extend_path
__path__ = extend_path(__path__, __name__)

Basically, it's a bit like the native way,
except that you keep your __init__.py modules,
but they must be identical across all packages!

The __init__.py modules of the plugin will overwrite the
__init__.py modules of the main package. If they are not
identical, things will break (mostly imports).

So, with the native way, you cannot write code in __init__.py
because those modules must not exist,
and with the pkgutil way you must not write code in __init__.py
because you don't want to have users duplicate it in their own.

At this state, I prefer the native way, as it's less files to write.

What I would have liked

Maybe namespace packages are not the best option for plugins.

But I would have loved being able to write my main package normally,
with __init__.py, and as much code in those as I want,
and write the plugin package without any __init__.py, so as to
"plug" (i.e. merge) additional modules into the main package:

# main package
package/
    __init__.py
    level1/
        __init__.py
        level2/
            __init__.py
            builtin_module.py

# plugin
package/
    level1/
        level2/
            plugin_module.py

The order of installation would not matter of course,
as merging the main package into the plugin one, or the opposite,
would result in the same final package on the disk:

# main package
package/
    __init__.py
    level1/
        __init__.py
        level2/
            __init__.py
            builtin_module.py
            plugin_module.py

The main package author doesn't even have to think about plugins
and prepare the field by removing __init__.py where it is needed.
Instead, they just write their package normally.

Then users can write a namespace package, mimicking the main package
structure, but without __init__.py files, and BOOM, they
successfully wrote a plugin!

Native namespace packages look like they are trying to be implicit,
but in my taste they are still not implicit enough.

Also, merging namespace packages into packages of the same name
would allow easy patching of projects! Something not quite working
like you want to? Quickly create a namespace package with just the
patched module, and list it as a dependency.

What do you use to allow plugins in your Python projects 🙂?

How to edit the contents of a git commit

Timothée Mazzucotelli — Mon, 05 Oct 2020 21:09:50 +0000

When you type change git commit or similar keywords in a search engine,
you find many answers explaning how to rewrite a commit message,
but not how to actually modify the contents of the commit.

This post quickly explains how to do that.

Lets begin by listing our commits:

In this screenshot I use my gl alias which expands to:

git log --pretty=format:'%C(yellow)%h %Cgreen%ad %Cblue%an%Cgreen%d %Creset%s' --date=short --graph

Here I want to modify the contents of the third commit,
feat: Refactor and add features, because I mistakenly
committed modifications to the changelog
which I didn't want.

So I run an interactive git rebase down to this commit
by running:

git rebase -i HEAD~3

This command launches your git editor (vim here)
to tell git what to do. Here we tell git to stop
right after the commit we want to modify:

We save and quit this temporary file (with :wq),
and git tells us that it stopped at the desired commit:

Don't pay attention to the commit SHA:
I took the screenshot afterwards so they don't match 😄

Now you can start modifying, adding or deleting files!

In my case I wanted to remove wrong sections in CHANGELOG.md,
as well as remove conflict-resolution lines in pyproject.toml.

In this screenshot I use my gs alias
which expands to git status -sb.

All you have to do now is to amend the current commit
(the one at which we stopped, the one we wanted to modify):

git commit -a --amend --no-edit
# --no-edit because we don't want to edit the message

And finally, let git finish the interactive rebase:

git rebase --continue

Done!

Django application as an authentication / authorization server for Shiny

Timothée Mazzucotelli — Fri, 25 Sep 2020 23:30:25 +0000

As you may know, Shiny Server comes in two versions: open-source and
professional. The professional adds security and authentication features
like password protected applications, and controlled access via SSL and LDAP,
Active Directory, Google OAuth, PAM, proxied authentication, or passwords.
If you need these authentication features but don't want or can't spend
$9,995 per year for the professional edition, then I got a solution for you!

In this post, I will show how to wrap one or several Shiny applications into
a Django application to add authentication and access control to your Shiny
apps. The shining star here will not be Django, as you could replace it by
any other web application you want, but the famous NginX reverse proxy,
accompanied by its devoted auth-request module.

The code used in this post is also available as a repository on GitHub.
It contains a Docker configuration so you can try it easily.

On the menu

Overview
Wrapping a Shiny app into a Django-powered page
Proxying Shiny requests to the Shiny app
Adding an authentication step for every Shiny request
Try it with a Dockerized project
Comments

What most of you will be interested in is starting at section 2:
Proxying, though Wrapping is interesting as well if you
want to create an interface to access multiple Shiny apps.

Overview

Let's look at some pictures to see what we want to accomplish. The first picture
shows our client-server architecture. The client can communicate with the server
on the port 80 or 443 (HTTP or HTTPs), but not on the ports 8000 or 8100, which
are used internally by Django and Shiny. This can be configured through the
firewall.

The second picture shows what happens when the client requests the URL
that the Shiny app is served on. As we wrap the Shiny app into a Django-powered
page, the request is proxied directly to Django by NginX. Django then gets the
initial Shiny page HTML contents with an internal HTTP Get request, and renders
it in a iframe (we will see the details later). It then returns this rendered
page to NginX, which returns it to the client.

The third picture shows each subsequent requests from the client to the server
through a WebSocket, and how NginX is asking authorization to Django. When NginX
receives the request, it sends a sub-request to Django, asking for permission
to proxy the request to Shiny and return the response to the client. If Django
says yes (HTTP 200), NginX proxies the request to Shiny. If Django says no
(HTTP 403 or any other error code), NginX rejects the request by returning
HTTP 403 as a response to the client.

OK, let's try it! To begin, create a directory that we will use for this
tutorial:

mkdir django-shiny
cd django-shiny

Wrapping a Shiny app into a Django-powered page

This first section will help you setup an example project to follow this
tutorial, but the first two steps described are optional. You can immediately
jump to the Proxying Shiny requests
section where we will a use pre-setup example project using Docker. You will
need to install Docker if you don't already have it. See Install Docker for
installation instructions.

The third step however might be interesting to read if you need to wrap your
Shiny app into a Django-powered page and website, instead of just using Django
as an external authentication / authorization backend.

The Shiny app
The Django app
Injecting the HTML contents in an Iframe

The Shiny app

Let's get a Shiny app example from RStudio's gallery.
The code is available on GitHub in this repository.

Clone it in a sub-directory called shinyapp:

git clone --depth=1 https://github.com/rstudio/shiny-examples
mv shiny-examples/001-hello shinyapp
rm -rf shiny-examples

We also need to install the Shiny R package. If you don't already have R
installed, you can install a recent version with the following commands:

sudo add-apt-repository "deb http://cran.rstudio.com/bin/linux/ubuntu trusty/"
sudo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys E084DAB9
sudo add-apt-repository ppa:marutter/rdev
sudo apt-get update
sudo apt-get install -y r-base

Run this command to install Shiny:

sudo R -e "install.packages('shiny', repos='https://cran.rstudio.com/')"

To run the Shiny application on port 8100, use the following command:

sudo R -e "shiny::runApp(appDir='shinyapp', port=8100)"

Try to go to http://localhost:8100 to see if the app is running.

The Django app

We will create a new Django project called djangoapp. If you don't already
have Django installed on your system, install it in a virtualenv with
pip install Django, or system-wide with sudo pip install Django.

To create the project, run the following command:

django-admin startproject djangoapp

We need to initialize the SQLite database first.

python djangoapp/manage.py migrate

You can now run the Django application on port 8000 with the following command:

python djangoapp/manage.py runserver localhost:8000

And try to go to http://localhost:8000 to see if the app is running.

Injecting the HTML contents in an Iframe

At this point you should have the following tree:

.
├── djangoapp
│   ├── db.sqlite3
│   ├── djangoapp
│   │   ├── __init__.py
│   │   ├── settings.py
│   │   ├── urls.py
│   │   └── wsgi.py
│   └── manage.py
└── shinyapp
    ├── app.R
    ├── DESCRIPTION
    └── Readme.md

3 directories, 9 files

We will proceed in three main steps:

Create a simple view that renders our wrapping HTML page
Create this wrapping HTML page that will:
1. Create an Iframe and add it to the DOM
2. Get the Shiny HTML contents in JSON format, with an asynchronous call
3. Inject these contents in the Iframe
Create the view to return Shiny HTML contents as JSON by:
1. Getting the contents with a GET request
2. Parsing it with BeautifulSoup's HTML parser
3. Dumping and returning it as JSON

Create the wrapping view

Let's create our wrapping view. First make sure you have listed djangoapp
in the Django settings' INSTALLED_APPS:

# settings.py

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'djangoapp'
]

Then we can add the URL in urls.py:

from django.contrib import admin
from django.urls import path

from . import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('shiny/', views.shiny, name='shiny'),
]

And now we create the view in a new views.py file:

touch djangoapp/djangoapp/views.py

from django.shortcuts import render, redirect

def shiny(request):
    return render(request, 'djangoapp/shiny.html')

Create the wrapping HTML page

Since we tell the view to render the djangoapp/shiny.html template, we need
to create it:

mkdir -p djangoapp/djangoapp/templates/djangoapp
touch djangoapp/djangoapp/templates/djangoapp/shiny.html
# too much of djangoapp already, I know

And write its contents. We simply add a title, to know we are in the wrapping
view, and then we add a script to retrieve the Shiny app contents:

<h1>We are in the wrapping page!</h1>

<div id="contents"></div>

<script src="https://code.jquery.com/jquery-3.3.1.min.js"></script>

<script>
  $(document).ready(function () {

    $.getJSON('{% url "shiny_contents" %}', function (data) {

      var iframe = document.createElement("iframe");
      $('#contents').append(iframe);

      iframe.contentWindow.document.open();
      iframe.contentWindow.document.write(data.html_contents);
      iframe.contentWindow.document.close();

      // Attempt circumvention
      if (iframe.contentWindow.WebSocket)
          WebSocket = iframe.contentWindow.WebSocket;
    });

  });

</script>

Several things happen here:

we declare a div of ID contents in which we will add an iframe,
we make use of JQuery's $(document).ready and $.getJSON methods to load HTML contents from an URL returning JSON,
we create the iframe, add it in the document, then write the HTML contents inside of it. We also reassign the WebSocket variable to the value of the iframe one.

Create the JSON view

As you maybe guessed, work is not finished. We need to add the shiny_contents
URL and view in the Django app. The view must return the contents of the Shiny
app initial page as JSON.

Add the URL in urls.py:

from django.contrib import admin
from django.urls import path

from . import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('shiny/', views.shiny, name='shiny'),
    path('shiny_contents/', views.shiny_contents, name='shiny_contents'),
]

Add the view to views.py:

from django.http import JsonResponse
from django.shortcuts import render

import requests

from bs4 import BeautifulSoup


def shiny(request):
    return render(request, 'djangoapp/shiny.html')


def shiny_contents(request):
    response = requests.get('http://localhost:8100')
    soup = BeautifulSoup(response.content, 'html.parser')
    return JsonResponse({'html_contents': str(soup)})

We are using requests and BeautifulSoup to get the HTML contents and return
it as text, dumped as JSON. If you know a better way, let me know in the
comments!

To install requests and BeautifulSoup: pip install requests beautifulsoup4.

OK, let's try! Run both the Shiny application and the Django application, then
go to http://localhost:8000/shiny to see how it goes!

You should see something like the following image:

But since Shiny requests are not proxied, they are simply lost in-between, and
your Shiny app will not respond, nor get a nice look because the static assets
will not be loaded. We will fix this in the next section:
Proxying Shiny requests.

This is it for the setup and the Django-wrapped Shiny page. For the rest of the
tutorial, I will explain how to configure NginX and Django to act as proxy
and authorization servers. The result is available for you to try in this
repository: https://github.com/Pawamoy/docker-nginx-auth-request-django-shiny-example.
See the Try it out section.

Proxying Shiny requests to the Shiny app

This is time for NginX to come on stage. We need to create our new,
project-specific configuration file in /etc/nginx/sites-available/djangoshiny.

We will tell NginX to proxy every request with an URL like shiny/* to our
Shiny app. All other requests will be proxied to the Django app.

# declare your Django app
upstream djangoapp_server {
    server localhost:8000;
}

# declare your Shiny app
upstream shinyapp_server {
    server localhost:8100;
}

# required for WebSockets
map $http_upgrade $connection_upgrade {
    default upgrade;
    '' close;
}

server {

    listen 80;
    server_name localhost;

    client_max_body_size 100M;

    # normal requests go to Django
    location / {
        proxy_pass http://djangoapp_server;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $host;
        proxy_redirect off;
        if (!-f $request_filename) {
            proxy_pass http://djangoapp_server;
            break;
        }
    }

    # "shiny" requests go to Shiny
    location ~ /shiny/.+ {

        rewrite ^/shiny/(.*)$ /$1 break;

        proxy_pass http://shinyapp_server;
        proxy_redirect http://shinyapp_server/ $scheme://$host/shiny/;

        # required for WebSockets
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;

        proxy_read_timeout 20d;
        proxy_buffering off;
    }

}

Once you're ready, enable this configuration by linking it like this:

sudo ln -s /etc/nginx/sites-available/djangoshiny /etc/nginx/sites-enabled/djangoshiny

Now reload NginX, and that's it! If you launch both Django on port 8000, and
Shiny on port 8100, you should be able to connect to http://localhost/shiny
and enjoy your Django-wrapped Shiny application functioning properly!

There is still one thing we need to add: authentication and authorization.
I won't show how to create sign-in / sign-up / sign-out views: you can
find all the documentation to enable Django authentication system online.
What I will show you is how to make Django act as an authorization server
for Shiny.

Adding an authentication step for every Shiny request

We want all requests proxied to Shiny to be authorized by Django. For this we
are gonna use the auth-request module. This module is not compiled in
NginX by default on distributions like Ubuntu / Debian. If you want to recompile
NginX with auth-request enabled, check how I do it in this Dockerfile.

An easier solution is to use the Docker setup from the same repository, with
the official NginX image which already supports auth-request.

So, once your NginX is ready, add this authorization step in the configuration
file:


    location ~ /shiny/.+ {

        # we tell nginx to call that location for each request
        auth_request /auth;

        rewrite ^/shiny/(.*)$ /$1 break;

        proxy_pass http://shinyapp_server;
        proxy_redirect http://shinyapp_server/ $scheme://$host/shiny/;

        # this part is needed for WebSockets to work
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;

        proxy_read_timeout 20d;
        proxy_buffering off;
    }

    # the /auth location will send a subrequest to django, URL shiny_auth/
    location = /auth {
        internal;

        proxy_pass http://localhost:8000/shiny_auth/;

        proxy_pass_request_body off;
        proxy_set_header Content-Length "";
        proxy_set_header X-Original-URI $request_uri;
    }

Of course, we also need to write the view called by the shiny_auth/ URL.
It's a very simple one. First add the URL in urls.py:

from django.contrib import admin
from django.urls import path

from . import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('shiny/', views.shiny, name='shiny'),
    path('shiny_contents/', views.shiny_contents, name='shiny_contents'),
    path('shiny_auth/', views.shiny_auth),
]

And then the view in views.py:

from django.http import HttpResponse

def shiny_auth(request):
    if request.user.is_authenticated:
        return HttpResponse(status=200)
    return HttpResponse(status=403)

Et voilà! If the user is authenticated, Django will say "OK". If the user is
not authenticated, Django will say "No". But you could implement any logic
you need instead of just checking if the user is authenticated or not. You
could have several Shiny apps served on different URLs, powered by Django
(in wrapping pages, as we saw earlier), and grant access to users with a
permission system, etc..

Try it with a Dockerized project

Clone the repository and build the images with the following commands:

git clone https://github.com/Pawamoy/docker-nginx-auth-request-django-shiny-example docker-django-shiny
cd docker-django-shiny
sudo make all

This first build should take quite some time. It will build Shiny and Django
images, get static assets, create the database, create a super user, and then
run the application. You will then be able to go to http://localhost:8000
to see it in action.

To print the available make commands, simply run make.

Docker Compose with NginX, Django, Gunicorn and multiple Postgres databases

Timothée Mazzucotelli — Fri, 25 Sep 2020 23:28:26 +0000

This post explains how to setup your Docker configuration for a web application
based on the Django framework. I got a lot of inspiration from other tutorials
and Docker examples: you can check these resources with the links at the bottom
of the post. You can also directly check the
repository
that reflects this tutorial.

In this particular example, we will use Gunicorn, but it should be easy enough
to replace it with an alternative Python WSGI HTTP server such as uwsgi. We will
also make use of pipenv, with related Pipfile and Pipfile.lock, instead of plain
pip and requirements.txt files.

Here is the plan:

Overview: to get a better understanding of the whole thing
Dockerfile: a simple Django application served by Gunicorn
Pipenv: spice things up with Pipfile and Pipfile.lock
Compose: add a container for NginX
Compose: add containers for one or more Postgres databases
Static files: collecting, storing and serving
Resources

Go to comments

Overview: to get a better understanding of the whole thing

So, let's start with some drawings in order to get a better idea of what we want
to accomplish, and how everything will fit together.

In this first, very simple image, you can see that we want three containers:
one for NginX, one for Django + Gunicorn (they always go together), and one for
our database. The NginX container communicate with the Django+Gunicorn one,
which itself connects to the Postgres container. Pretty straight-forward, right?

In our configuration, it means we will declare three containers, or three
services if we talk in terms of Docker Compose.

Except that we need bridges between the containers, in order for them to
communicate. Let's add these bridges:

In docker-compose.yml, we will declare these bridges thanks to the networks
directive, and connect them to the right containers.

Of course, you may want or need several databases for your project. So here is
an updated image with two database containers. It's simply a matter of adding
a new brige:

Once you know how to do it for two databases, it's very easy to add more.

Now, this is enough for local development. But each time you restart your
containers or services, the data in the Postgres databases will be lost. In
production, we need these data to be persistent. If we keep the data in
production, let's keep them in local environment as well. To do this, we will
use volumes, a feature of Docker:

Alright, that is enough for the overview, let's get our hands dirty!

Dockerfile: a simple Django application served by Gunicorn

If you don't already have a simple Django project available for testing,
I invite you to create one with django-admin startproject hello.

Here is the directory/file tree you should have in order to follow this
tutorial:

.                         # Your current directory, created for this tutorial
└── hello                 # The Django project
    ├── hello             # The main Django app of your project
    │   ├── __init__.py
    │   ├── settings.py
    │   ├── urls.py
    │   └── wsgi.py
    └── manage.py

Now that you have a working Django project, you can run it by going into the
hello directory and type ./manage.py runserver. Go to
http://localhost:8000 to see the result.

Instead of running it with the Django runserver management command, let's try
with Gunicorn. First, install it with pip install gunicorn, be it in a
virtualenv or system-wide with sudo pip install gunicorn.

It's as easy as running gunicorn --bind :8000 hello.wsgi:application from
inside the Django project. If you are one directory above, use
gunicorn --chdir hello --bind :8000 hello.wsgi:application.

We have all we need to write our Dockerfile:

# start from an official image
FROM python:3.6

# arbitrary location choice: you can change the directory
RUN mkdir -p /opt/services/djangoapp/src
WORKDIR /opt/services/djangoapp/src

# install our two dependencies
RUN pip install gunicorn django

# copy our project code
COPY . /opt/services/djangoapp/src

# expose the port 8000
EXPOSE 8000

# define the default command to run when starting the container
CMD ["gunicorn", "--chdir", "hello", "--bind", ":8000", "hello.wsgi:application"]

The Dockerfile must be placed at the root of your test directory. As a reminder:

.                         # Your current directory, created for this tutorial
├── hello                 # The Django project
│   ├── hello             # The main Django app of your project
│   └── manage.py
└── Dockerfile            # Your Dockerfile

We are now able to build our container with docker build . -t hello, and to
start it with docker run -p 8000:8000 hello. The -p 8000:8000 option says to
bind the port 8000 of the host to the port 8000 of the container, allowing you
to go to http://localhost:8000 and see your application
running as if you were inside of the container.

Pipenv: spice things up with Pipfile and Pipfile.lock

This step is completely optional. If you prefer to use plain pip and
requirements files, you can skip this section.

First install pipenv with pip install pipenv, or system-wide with
sudo pip install pipenv.
Since we only need Django and Gunicorn, our Pipfile will be very simple:

[[source]]
url = "https://pypi.python.org/simple"
verify_ssl = true
name = "pypi"


[packages]
Django = "*"
gunicorn = "*"


[requires]
# our Dockerfile is based on Python 3.6
python_version = "3.6"

Just like the Dockerfile, Pipfile must be placed at the root of the project.

.
├── hello
│   ├── hello
│   └── manage.py
├── Dockerfile
└── Pipfile

Simply run pipenv lock to create Pipfile.lock from Pipfile.

Now we need to update our Dockerfile to use pipenv:

# start from an official image
FROM python:3.6

# arbitrary location choice: you can change the directory
RUN mkdir -p /opt/services/djangoapp/src
WORKDIR /opt/services/djangoapp/src

# install our dependencies
# we use --system flag because we don't need an extra virtualenv
COPY Pipfile Pipfile.lock /opt/services/djangoapp/src/
RUN pip install pipenv && pipenv install --system

# copy our project code
COPY . /opt/services/djangoapp/src

# expose the port 8000
EXPOSE 8000

# define the default command to run when starting the container
CMD ["gunicorn", "--chdir", "hello", "--bind", ":8000", "hello.wsgi:application"]

You can rebuild the image with docker build . -t hello and try to run it again
to see if everything works correctly.

Compose: add a container for NginX

Since we will then have two containers, one for Django + Gunicorn, and one for
NginX, it's time to start our composition with Docker Compose and
docker-compose.yml. Create your docker-compose.yml file at the root of the
project, like following:

.
├── hello
│   ├── hello
│   └── manage.py
├── docker-compose.yml
├── Dockerfile
└── Pipfile

We are gonna use the version 3 of the configuration syntax. First, we add
the Django+Gunicorn service:

version: '3'

services:
  djangoapp:
    build: .
    volumes:
      - .:/opt/services/djangoapp/src
    ports:
      - 8000:8000

We simply tell Docker Compose that the djangoapp service must use an image
that is built from the current directory, therefore looking for our Dockerfile.
The volumes directive tells to bind the current directory of the host to the
/opt/services/djangoapp/src directory of the container. The changes in our
current directory will be reflected in real-time in the container directory.
And reciprocally, changes that occur in the container directory will occur
in our current directory as well.

Build and run the service with docker-compose up. The name of the image will
be automatically chosen by Docker Compose (it will be the name of the current
directory with _djangoapp appended).

Ok, let's add our NginX service now:

version: '3'

services:

  djangoapp:
    build: .
    volumes:
      - .:/opt/services/djangoapp/src

  nginx:
    image: nginx:1.13
    ports:
      - 8000:80
    volumes:
      - ./config/nginx/conf.d:/etc/nginx/conf.d
    depends_on:  # <-- wait for djangoapp to be "ready" before starting this service
      - djangoapp

Note that we removed the ports directive from our djangoapp service. Indeed
we will not communicate directly with Gunicorn anymore, but with NginX. We still
want to access our app at http://localhost:8000, and
we want NginX to listen to the port 80 in the container, so we use
ports: - 8000:80.

Note: in a production environment, we would use 80:80 instead.

We also bind a local directory to the /etc/nginx/conf.d container directory.
Let's create it and see what's inside:

mkdir -p config/nginx/conf.d
touch config/nginx/conf.d/local.conf

You should now have the following files and directories:

.
├── config
│   └── nginx
│       └── conf.d
│           └── local.conf
├── docker-compose.yml
├── Dockerfile
├── hello
│   ├── hello
│   └── manage.py
├── Pipfile
└── Pipfile.lock

The config/nginx/conf.d/local.conf file contains our NginX configuration:

# first we declare our upstream server, which is our Gunicorn application
upstream hello_server {
    # docker will automatically resolve this to the correct address
    # because we use the same name as the service: "djangoapp"
    server djangoapp:8000;
}

# now we declare our main server
server {

    listen 80;
    server_name localhost;

    location / {
        # everything is passed to Gunicorn
        proxy_pass http://hello_server;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $host;
        proxy_redirect off;
    }
}

But before we try this out, remember that we need a bridge to make our services
able to communicate?

Update your docker-compose.yml as follow:

version: '3'

services:

  djangoapp:
    build: .
    volumes:
      - .:/opt/services/djangoapp/src
    networks:  # <-- here
      - nginx_network

  nginx:
    image: nginx:1.13
    ports:
      - 8000:80
    volumes:
      - ./config/nginx/conf.d:/etc/nginx/conf.d
    depends_on:
      - djangoapp
    networks:  # <-- here
      - nginx_network

networks:  # <-- and here
  nginx_network:
    driver: bridge

Run docker-compose up and see if you can still see the Django default page
at http://localhost:8000.

Compose: add containers for one or more Postgres databases

We now want to use Postgres instead of the starting default SQLite database.
We will need to update several things: our Pipfile, because we need the
psycopg2 Python package, the Postgres driver; our Django project settings; and
our docker-compose.yml file.

Pipfile becomes:

  [[source]]
  url = "https://pypi.python.org/simple"
  verify_ssl = true
  name = "pypi"


  [packages]
  Django = "*"
  gunicorn = "*"
  "psycopg2" = "*"


  [requires]
  # our Dockerfile is based on Python 3.6
  python_version = "3.6"

Don't forget to run pipenv lock to update your lock file, and rebuild your
Docker image with docker-compose build.

In the Django project settings, update the DATABASE setting from:

  DATABASES = {
      'default': {
          'ENGINE': 'django.db.backends.sqlite3',
          'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
      }
  }

...to:

  DATABASES = {
      'default': {
          'ENGINE': 'django.db.backends.postgresql_psycopg2',
          'NAME': 'database1',
          'USER': 'database1_role',
          'PASSWORD': 'database1_password',
          'HOST': 'database1',  # <-- IMPORTANT: same name as docker-compose service!
          'PORT': '5432',
      }
  }

As you can see, we used database1 everywhere, for the name, user, password
and host. In fact, we can change these values to whatever suits us. But we
must ensure the database container will use the same values! To do that, we
will copy these values in a configuration file destined to be read by our
database container.

Create a db directory in the config one, and add the database1_env file:

  mkdir config/db
  touch config/db/database1_env

The contents of config/db/database1_env must then be:

  POSTGRES_USER=database1_role
  POSTGRES_PASSWORD=database1_password
  POSTGRES_DB=database1

These variable are used by the Postgres Docker image, for more information
please check out the documentation on
docs.docker.com
or hub.docker.com.

It means that, when started, the Postgres container will create a database
called database1, assigned to the role database1_role with password
database1_password. If you change these values, remember to also change them
in the DATABASES setting.

We are now ready to add our service in docker-compose.yml. The added service must have the same name than what is declared in the DATABASES setting:

  version: '3'

  services:

    djangoapp:
      build: .
      volumes:
        - .:/opt/services/djangoapp/src
      networks:
        - nginx_network
        - database1_network  # <-- connect to the bridge
      depends_on:  # <-- wait for db to be "ready" before starting the app
        - database1

    nginx:
      image: nginx:1.13
      ports:
        - 8000:80
      volumes:
        - ./config/nginx/conf.d:/etc/nginx/conf.d
      depends_on:
        - djangoapp
      networks:
        - nginx_network

    database1:  # <-- IMPORTANT: same name as in DATABASES setting, otherwise Django won't find the database!
      image: postgres:10
      env_file:  # <-- we use the previously defined values
        - config/db/database1_env
      networks:  # <-- connect to the bridge
        - database1_network
      volumes:
        - database1_volume:/var/lib/postgresql/data

  networks:
    nginx_network:
      driver: bridge
    database1_network:  # <-- add the bridge
      driver: bridge

  volumes:
    database1_volume:

You should be able to understand everything here. However, we added two new
things: the database1: volumes: directive, and the root volumes:
directive. You need to declare your volumes in the root volumes: directive
if you want them to be kept persistently. Then, you can bind a volume to a
directory in the container. Here, we bind our declared database1_volume to
the database1 container's /var/lib/postgresql/data directory. Everything
added to this directory will be persistently stored in the volume called
database1_volume. So each subsequent run of the container will have access
to the previous data! It means you can stop and restart your service without
losing the data.

OK, let's try it. As we are using Django, we need to "migrate" the database
first. To do this, we will simply use Docker Compose to start our djangoapp
service and run the migration command inside it:

docker-compose build  # to make sure everything is up-to-date
docker-compose run --rm djangoapp /bin/bash -c "cd hello; ./manage.py migrate"

From now on, it should be really easy to add other databases: just add other
database services (database2) with their networks volumes (remember to
connect the networks and bind the volumes), update your DATABASES setting in
the Django project, and create the environment file for each database in
config/db.

Static files: collecting, storing and serving

UPDATE: a lot of us (including myself) seem to have trouble updating
the static files in the designated Docker volume. In retrospect,
collecting the static files in the Dockerfile might not be a good idea,
because of the caching mechanisms of Docker. A simple and effective solution,
mentioned by Isaac T Chikutukutu in the comments, is to remove this command
from the Dockerfile and manually run it when needed:
docker-compose run djangoapp hello/manage.py collectstatic --no-input

I edited the rest of the post accordingly.

(End of update)

Let's not forget about the static files! In order for NginX to serve them,
we will update the config/nginx/conf.d/local.conf file, as well as our
docker-compose.yml file. Static files will be stored in volumes.
We also need to set the STATIC_ROOT and MEDIA_ROOT variables
in the Django project settings.

NginX configuration:

  upstream hello_server {
      server djangoapp:8000;
  }

  server {

      listen 80;
      server_name localhost;

      location / {
          proxy_pass http://hello_server;
          proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
          proxy_set_header Host $host;
          proxy_redirect off;
      }

      location /static/ {
          alias /opt/services/djangoapp/static/;
      }

      location /media/ {
          alias /opt/services/djangoapp/media/;
      }
  }

Django project settings:

  # as declared in NginX conf, it must match /opt/services/djangoapp/static/
  STATIC_ROOT = os.path.join(os.path.dirname(os.path.dirname(BASE_DIR)), 'static')

  # do the same for media files, it must match /opt/services/djangoapp/media/
  MEDIA_ROOT = os.path.join(os.path.dirname(os.path.dirname(BASE_DIR)), 'media')

Volumes in docker-compose.yml:

  version: '3'

  services:

    djangoapp:
      build: .
      volumes:
        - .:/opt/services/djangoapp/src
        - static_volume:/opt/services/djangoapp/static  # <-- bind the static volume
        - media_volume:/opt/services/djangoapp/media  # <-- bind the media volume
      networks:
        - nginx_network
        - database1_network
      depends_on:
        - database1

    nginx:
      image: nginx:1.13
      ports:
        - 8000:80
      volumes:
        - ./config/nginx/conf.d:/etc/nginx/conf.d
        - static_volume:/opt/services/djangoapp/static  # <-- bind the static volume
        - media_volume:/opt/services/djangoapp/media  # <-- bind the media volume
      depends_on:
        - djangoapp
      networks:
        - nginx_network

    database1:
      image: postgres:10
      env_file:
        - config/db/database1_env
      networks:
        - database1_network
      volumes:
        - database1_volume:/var/lib/postgresql/data

  networks:
    nginx_network:
      driver: bridge
    database1_network:
      driver: bridge

  volumes:
    database1_volume:
    static_volume:  # <-- declare the static volume
    media_volume:  # <-- declare the media volume

And finally, collect the static files each time you need to update them, by running:

  docker-compose run djangoapp hello/manage.py collectstatic --no-input

Resources

Here are the resources I used to write this tutorial:

And here is the
repository
that reflects this tutorial (with a few more things).

Don't hesitate to share other interesting resources in the comment section!

Migrate Disqus comments to Utterances (in GitHub issues) with Python

Timothée Mazzucotelli — Fri, 25 Sep 2020 23:24:20 +0000

When I replaced Jekyll
and my Jekyll-ReadTheDocs theme
with MkDocs
and a blog-customised version of
the Material for MkDocs theme,
the URLs of my posts changed.

I was using Disqus for comments,
and they provide a way to migrate threads from old URLs to new URLs.
Unfortunately, this time it didn't work for some reason
(I had already done it once in the past and it worked fine).

I've read more and more criticism about Disqus related to privacy,
so I looked at a replacement.
The Disqus thread migration was not working so it was the perfect occasion!

I've read a few webpages and got interested in Isso.
Unfortunately again, I did not manage
to install it on my Raspberry Pi.

So I went with a much simpler solution: Utterances.
You basically enable a GitHub app on your repository,
add a script in your posts pages, and voilà: your new comment section
powered by GitHub issues.

I'm not completely satisfied because readers will need to log in
with their GitHub account to comment (no anonymous comments),
and you can't have nested discussions (well, just like in GitHub issues).

But this was really easy to setup 😄!

Now I just needed to migrate the Disqus comments into GitHub issues.
To do this semi-automatically, I wrote the following script.
"Semi-automatically" because I still had to write a "test" comment
in each one of my posts so Utterances would initiate/create the issues.
I guess it could be implemented in the script directly,
but since I just had a dozen of posts, I took the easy/repetitive way.

The script

!!! warning
You'll have to create the initial issues before running the script!
Serve your blog locally, and write a "test" comment in every post that has comments.

First, export your Disqus comments.

Then you'll need two Python libraries:

You'll also need to create a token on GitHub,
with just the public repos access.

Write the following environment variables in a file,
and source it.

export FILEPATH="comments.xml"
export USERNAME="yourUsername"
export TOKEN="yourToken"
export REPOSITORY="yourUsername/yourUsername.github.io"
export BASE_URL="https://yourUsername.github.io/"

Now you can copy/paste and run this script:

import xmltodict
from github import Github

FILEPATH = os.environ["FILEPATH"]
USERNAME = os.environ["USERNAME"]
TOKEN = os.environ["TOKEN"]
REPOSITORY = os.environ["REPOSITORY"]
BASE_URL = os.environ["BASE_URL"]


def disqus_to_github():
    g = Github(TOKEN)
    repo = g.get_repo(REPOSITORY)
    issues = repo.get_issues()

    with open(FILEPATH) as fd:
        data = xmltodict.parse(fd.read())

    data = data["disqus"]

    threads = [dict(t) for t in data["thread"]]
    posts = sorted((dict(p) for p in data["post"]), key=lambda d: d["createdAt"])

    # only keep threads with comments
    twc_ids = set(p["thread"]["@dsq:id"] for p in posts)
    threads = {t["@dsq:id"]: t for t in threads if t["@dsq:id"] in twc_ids}

    # associate the thread to each post
    for post in posts:
        post["thread"] = threads[post["thread"]["@dsq:id"]]

    # associate the related GitHub issue to each thread
    # warning: the issues need to exist before you run this script!
    # write a "test" comment in each one of your post with comments
    # to make Utterances create the initial issues
    for thread in threads.values():
        for issue in issues:
            if issue.title == thread["link"].replace(BASE_URL, ""):
                thread["issue"] = issue
                break

    # iterate on posts and create issues comments accordingly
    for i, post in enumerate(posts, 1):
        name = post["author"]["name"]
        user = post["author"].get("username")
        mention = " @" + user if user and not user.startswith("disqus_") else ""
        date = post["createdAt"]
        message = post["message"]
        issue = post["thread"]["issue"]
        body = f"*Original date: {date}*\n\n{message}"
        # don't add original author when it's you
        if user != USERNAME:
            body = f"*Original author:* **{name}{mention}**  \n{body}" 
        print(f"Posting {i}/{len(posts)} to issue {issue.number}    \r", end="")
        issue.create_comment(body)

    print()


if __name__ == "__main__":
    disqus_to_github()

I wrote this for a one-time, personal use only,
so it could easily crash when you try it!
Just use your Python skills and adapt it 😉

Unify Python logging for a Gunicorn/Uvicorn/FastAPI application

Timothée Mazzucotelli — Fri, 25 Sep 2020 16:42:07 +0000

I recently started playing with FastAPI
and HTTPX,
and I am deploying my app with Gunicorn and
Uvicorn workers.

But when serving, the logs from each component looks quite different
from the others. I want them to all look the same, so I can easily read them
or exploit them in something like Kibana.

After a lot of hours trying to understand
how Python logging works,
and how to override libraries' logging settings,
here is what I have...

A single run.py file!
I didn't want to split logging configuration, Gunicorn configuration,
and the rest of the code into multiple files, as it was harder to wrap
my head around it.

Everything is contained in this single file:

import os
import logging
import sys

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger

from my_app.app import app


LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False
WORKERS = int(os.environ.get("GUNICORN_WORKERS", "5"))


class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


class StubbedGunicornLogger(Logger):
    def setup(self, cfg):
        handler = logging.NullHandler()
        self.error_logger = logging.getLogger("gunicorn.error")
        self.error_logger.addHandler(handler)
        self.access_logger = logging.getLogger("gunicorn.access")
        self.access_logger.addHandler(handler)
        self.error_log.setLevel(LOG_LEVEL)
        self.access_log.setLevel(LOG_LEVEL)


class StandaloneApplication(BaseApplication):
    """Our Gunicorn application."""

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == '__main__':
    intercept_handler = InterceptHandler()
    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
    # logging.root.handlers = [intercept_handler]
    logging.root.setLevel(LOG_LEVEL)

    seen = set()
    for name in [
        *logging.root.manager.loggerDict.keys(),
        "gunicorn",
        "gunicorn.access",
        "gunicorn.error",
        "uvicorn",
        "uvicorn.access",
        "uvicorn.error",
    ]:
        if name not in seen:
            seen.add(name.split(".")[0])
            logging.getLogger(name).handlers = [intercept_handler]

    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])

    options = {
        "bind": "0.0.0.0",
        "workers": WORKERS,
        "accesslog": "-",
        "errorlog": "-",
        "worker_class": "uvicorn.workers.UvicornWorker",
        "logger_class": StubbedGunicornLogger
    }

    StandaloneApplication(app, options).run()

If you are in a hurry, copy-paste it, change the Gunicorn options at the end,
and try it!

If you're not, I will explain each part below.

import os
import logging
import sys

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger

This part is easy, we simply import the things we need.
The Gunicorn BaseApplication so we can run Gunicorn directly from this script,
and its Logger that we will override a bit.
We are using Loguru later in the code,
to have a pretty log format, or to serialize them.

from my_app.app import app

In my project, I have a my_app package with an app module.
My FastAPI application is declared in this module, something like
app = FastAPI().

LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False
WORKERS = int(os.environ.get("GUNICORN_WORKERS", "5"))

We setup some values from environment variables, useful for development vs.
production setups. JSON_LOGS tells if we should serialize the logs to JSON,
and WORKERS tells how many workers we want to have.

class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

This code is copy-pasted from Loguru's documentation!
This handler will be used to intercept the logs emitted by libraries
and re-emit them through Loguru.

class StubbedGunicornLogger(Logger):
    def setup(self, cfg):
        handler = logging.NullHandler()
        self.error_logger = logging.getLogger("gunicorn.error")
        self.error_logger.addHandler(handler)
        self.access_logger = logging.getLogger("gunicorn.access")
        self.access_logger.addHandler(handler)
        self.error_log.setLevel(LOG_LEVEL)
        self.access_log.setLevel(LOG_LEVEL)

This code was copied from this
GitHub comment
by @dcosson. Thanks!
It will allow us to override Gunicorn's own logging configuration
so its logs can be formatted like the rest.

I'm not sure about the last two lines, as removing them doesn't change anything.
There are still mysteries about Python logging that I couldn't resolve...

class StandaloneApplication(BaseApplication):
    """Our Gunicorn application."""

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

This code is taken from
Gunicorn's documentation.
We declare a simple Gunicorn application that we will be able to run.
It accepts all Gunicorn's options.

if __name__ == '__main__':
    intercept_handler = InterceptHandler()
    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
    # logging.root.handlers = [intercept_handler]
    logging.root.setLevel(LOG_LEVEL)

We simply instantiate our interception handler,
and set the log level on the root logger.

Once again, I fail to understand how this works exactly, as the two commented
lines have no impact on the result. I did a lot of trial and error and ended
up with something working, but I cannot entirely explain why.
The idea here was to set the handler on the root logger so it intercepts
everything, but it was not enough (logs were not all intercepted).

    seen = set()
    for name in [
        *logging.root.manager.loggerDict.keys(),
        "gunicorn",
        "gunicorn.access",
        "gunicorn.error",
        "uvicorn",
        "uvicorn.access",
        "uvicorn.error",
    ]:
        if name not in seen:
            seen.add(name.split(".")[0])
            logging.getLogger(name).handlers = [intercept_handler]

Here we iterate on all the possible loggers declared by libraries
to override their handlers with our interception handler.
This is where we actually configure every logger to behave the same.

For a reason that I fail to understand, Gunicorn and Uvicorn do not appear
in the root logger manager, so we have to hardcode them in the list.

We also use a set to avoid setting the interception handler on the parent
of a logger that is already configured, because otherwise logs would be
emitted twice or more. I'm not sure this code can handle levels of
nested loggers deeper than two.

    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])

Here we configure Loguru to write on the standard output,
and to serialize logs if needed.

At some point I was also using activation=[("", True)]
(see Loguru's docs),
but it seems it's not required either.

    options = {
        "bind": "0.0.0.0",
        "workers": WORKERS,
        "accesslog": "-",
        "errorlog": "-",
        "worker_class": "uvicorn.workers.UvicornWorker",
        "logger_class": StubbedGunicornLogger
    }

    StandaloneApplication(app, options).run()

Finally, we set our Gunicorn options, wiring things up,
and run our application!

Well, I'm not really proud of this code, but it works!

Forem: Timothée Mazzucotelli

Testing a FastAPI application using Ormar models and Alembic migrations

Database models

Database and migrations helpers

Automatic database creation/update

Pytest fixture

Model instances factories

Populating the database with data

Example tests

How to add Alembic migrations to an existing FastAPI + Ormar project

Local HTTP server sending fake files for testing purposes

The basics: http.server

Serving files from memory

Multi-threading

Reusable address

Still not enough

With FastAPI

Trying to be clever

Additional niceties

Pytest fixture

Passing Makefile arguments to a command, as they are typed in the command line

Experiments

Sorcery

Recipe

Side-effects

Addendum

Python packages and plugins as namespace packages

Native namespace packages

pkgutil-style namespace packages

What I would have liked

How to edit the contents of a git commit

Django application as an authentication / authorization server for Shiny

Overview

Wrapping a Shiny app into a Django-powered page

The Shiny app

The Django app

Injecting the HTML contents in an Iframe

Create the wrapping view

Create the wrapping HTML page

Create the JSON view

Proxying Shiny requests to the Shiny app

Adding an authentication step for every Shiny request

Try it with a Dockerized project

Docker Compose with NginX, Django, Gunicorn and multiple Postgres databases

Overview: to get a better understanding of the whole thing

Dockerfile: a simple Django application served by Gunicorn

Pipenv: spice things up with Pipfile and Pipfile.lock

Compose: add a container for NginX

Compose: add containers for one or more Postgres databases

Static files: collecting, storing and serving

Resources

Migrate Disqus comments to Utterances (in GitHub issues) with Python

The script

Unify Python logging for a Gunicorn/Uvicorn/FastAPI application

The basics: `http.server`

`pkgutil`-style namespace packages