Introduction aka "all flatterers live thanks to those who listen to them"

If you're not already in love with FastAPI then it's just a matter of time, the documentation is amazing, the community on gitter is friendly and helping, and the creator tiangolo simply amazing.

Now the famous Jean de la Fontaine poem says:

all flatterers live thanks to those who listen to them

So let's keep the praise short and try to bring one (or two) interesting things to the table (we hope):

  1. a use-case of a recursive pydantic model (under the hood FastAPI uses pydantic which is equally amazing) using forward references
  2. how to implement a small and simple logging utility that let you change debugging levels on the fly in your application
Le Corbeau et le Renard

Some background

I always found logging was a rather unfriendly topic in Python, and while developing web applications you end up often having too many logs, or not enough (couldn't help making a meme...).

print vs logging

The little snippets below will show how to bring into your API the ability to modify the log levels of a defined logger on the fly, while it's running.

Create a self-referencing recursive model

I won't explain badly how to use the logging module in python, some people before me did it very well (for instance here and here).

The main thing is that the various loggers can be represented as a tree, with the root logger at the top.

This post written by the creator of logging_tree package (this package let you introspect the various loggers) explains it well.

A pydantic model that can be used to represent such a tree could looks like:

from __future__ import annotations
from typing import List, Optional

from pydantic import BaseModel


class LoggerModel(BaseModel):
    name: str
    level: Optional[int]
    children: Optional[List[LoggerModel]] = None


LoggerModel.update_forward_refs()

We represent in our app the logging tree with this LoggerModel object which gets a name, a level and childrens, those being themselves LoggerModel objects !

As at the time of initialization the children list of subtree doesn't exist, so one can use postponed annotations with the help of the future statement:

from __future__ import annotations

We also need to use pydantic update_forward_refs() method to take into account the self-referencing nature of the model.

Access and use this LoggerModel in FastAPI

First we have a simple route to list all the loggers of the logging tree

@router.get("/", response_model=LoggerModel)
def loggers_list():
    rootm = generate_tree()
    logger.debug(rootm)
    return rootm

the generate_tree function looks like this:

def generate_tree() -> LoggerModel:
    # adapted from logging_tree package https://github.com/brandon-rhodes/logging_tree
    rootm = LoggerModel(
        name="root", level=logging.getLogger().getEffectiveLevel(), children=[]
    )
    nodesm = {}
    items = list(logging.root.manager.loggerDict.items())  # type: ignore
    items.sort()
    for name, loggeritem in items:
        if isinstance(loggeritem, logging.PlaceHolder):
            nodesm[name] = nodem = LoggerModel(name=name, children=[])
        else:
            nodesm[name] = nodem = LoggerModel(
                name=name, level=loggeritem.getEffectiveLevel(), children=[]
            )
        i = name.rfind(".", 0, len(name) - 1)  # same formula used in `logging`
        if i == -1:
            parentm = rootm
        else:
            parentm = nodesm[name[:i]]
        parentm.children.append(nodem)
    return rootm

It simply builds the tree of loggers into the LoggerModel object described above, iterating through the nodes.

If you open the auto-generated OpenAPI documentation, you'll get this, in particular the response_model=LoggerModel is displayed so you know what kind of answer to expect from the API:

OpenAPI docs

You can also display a logger info, with this other route :

@router.get("/{logger_name}", response_model=LoggerModel)
def logger_get(logger_name: str):
    logger.debug(f"getting logger {logger_name}")
    rootm = generate_tree()
    lm = get_lm_from_tree(rootm, logger_name)
    if lm is None:
        raise HTTPException(status_code=404, detail=f"Logger {logger_name} not found")
    return lm

With the get_lm_fron_tree being that recursive function:

def get_lm_from_tree(loggertree: LoggerModel, find_me: str) -> LoggerModel:
    if find_me == loggertree.name:
        logger.debug("Found")
        return loggertree
    else:
        for ch in loggertree.children:
            logger.debug(f"Looking in: {ch.name}")
            i = get_lm_from_tree(ch, find_me)
            if i:
                return i

So for instance my application has one routers module with two files loggers.py and users.py, each one having a logger = logging.getLogger(__name__) defined at the top.

If you want to have the info about that logger, you hit the route with its name and get this:

routers loggers display

Each child is set to logging level 20, that's INFO level, and none has children.

The parent has no logger set hence the null level.

And finally we create a patch route, that enables you to change the logging level of whatever logger you want:

@router.patch("/")
def logger_patch(loggerpatch: LoggerPatch):
    rootm = generate_tree()
    lm = get_lm_from_tree(rootm, loggerpatch.name)
    logger.debug(f"Actual level of {lm.name} is {lm.level}")
    logger.debug(f"Setting {loggerpatch.name} to {loggerpatch.level}")
    logging.getLogger(loggerpatch.name).setLevel(LOG_LEVELS[loggerpatch.level])
    return loggerpatch

A use case while your app is running

So now let's say you implemented the snippets above and you have your app running fine until you suspect something is happening in let's say the databases package.

You have a patch route to change a user name/email/password of your users in database, and it is not doing what you thought it would, you'd like to see what query it sends.

The following line in database is what spits out the query sent:

logger.debug("Query: %s\nArgs: %s", compiled_query, args)

Depending on how you started your app logging you may just have gunicorn logs.

And unless you explicitly said that you wanted databases package logs to be set on debug you won't see them.

You can :

  1. stop your app, change the logging level of the databases package, restart it
  2. OR : you can use the loggers patch route defined above, set the logging level of databases to debug

In the log below you can see an example. The 1st time a user is patched you just have gunicorn this gunicorn log:

2019-06-10 08:12:30,480 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "PATCH /users/string HTTP/1.1" 204

Then databases package logger is patched to spit out debug logs:

2019-06-10 08:12:43,395 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "PATCH /loggers/ HTTP/1.1" 200

And finally the very same patch query on the users route is sent, and you now see what databases passed effectively as a query and args:

2019-06-10 08:12:57,098 - databases - 194 - DEBUG - Query: SELECT users.id, users.name, users.email, users.password, users.time_created, users.time_updated 
FROM users 
WHERE users.name = $1
Args: ['string']
2019-06-10 08:12:57,178 - databases - 194 - DEBUG - Query: UPDATE users SET name=$2, email=$1, password=$4, time_updated=now() WHERE users.name = $3
Args: ['user@example.com', 'string', 'string', '$argon2i$v=19$m=102400,t=2,p=8$l5JyrlUK4XxPqfUeI0RISQ$NsIUwxZk1Bd791AGe7BSWw']

see the whole logs for that little interaction just here:

2019-06-10 08:12:10,966 - gunicorn.error - 271 - INFO - Starting gunicorn 19.9.0
2019-06-10 08:12:10,967 - gunicorn.error - 271 - INFO - Listening at: http://0.0.0.0:8000 (1)
2019-06-10 08:12:10,967 - gunicorn.error - 271 - INFO - Using worker: uvicorn.workers.UvicornWorker
2019-06-10 08:12:10,970 - gunicorn.error - 271 - INFO - Booting worker with pid: 11
2019-06-10 08:12:11,020 - gunicorn.error - 271 - INFO - Booting worker with pid: 12
2019-06-10 08:12:11,431 - gunicorn.error - 271 - INFO - Started server process [11]
2019-06-10 08:12:11,431 - gunicorn.error - 271 - INFO - Waiting for application startup.
2019-06-10 08:12:11,498 - gunicorn.error - 271 - INFO - Started server process [12]
2019-06-10 08:12:11,498 - gunicorn.error - 271 - INFO - Waiting for application startup.
2019-06-10 08:12:14,098 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "GET /docs HTTP/1.1" 200
2019-06-10 08:12:14,304 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "GET /openapi.json HTTP/1.1" 200
2019-06-10 08:12:30,480 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "PATCH /users/string HTTP/1.1" 204
2019-06-10 08:12:43,395 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "PATCH /loggers/ HTTP/1.1" 200
2019-06-10 08:12:48,304 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "GET /loggers/ HTTP/1.1" 200
2019-06-10 08:12:57,098 - databases - 194 - DEBUG - Query: SELECT users.id, users.name, users.email, users.password, users.time_created, users.time_updated 
FROM users 
WHERE users.name = $1
Args: ['string']
2019-06-10 08:12:57,178 - databases - 194 - DEBUG - Query: UPDATE users SET name=$2, email=$1, password=$4, time_updated=now() WHERE users.name = $3
Args: ['user@example.com', 'string', 'string', '$argon2i$v=19$m=102400,t=2,p=8$l5JyrlUK4XxPqfUeI0RISQ$NsIUwxZk1Bd791AGe7BSWw']
2019-06-10 08:12:57,182 - gunicorn.error - 271 - INFO - ('192.168.112.1', 39190) - "PATCH /users/string HTTP/1.1" 204

Conclusion

As you can see, thanks to FastAPI and pydantic it's very simple to create what could seem at first complicate self-referencing, recursive models.

A further improvement will be to be able to change handlers on the fly thus enabling you to log to files, console and change it at will.