FastAPI logging in routes in Azure Functions

Question

I noticed that any logging.info('stuff happened in a route') when in a route, it wasn't coming through to my Application Insights log. The instances of logging.info('stuff happened in entrypoint') were. I found this bit of advice but the solution is a bit vague and seemingly specific to running uvicorn and I don't know how to apply it to Azure.

For reference here are snippets from my __init__.py

import logging
import azure.functions as func
from fastapi import FastAPI
from .routes import route1

app = FastAPI()
app.include_router(route1.router)

@app.get("/api/test"):
def test():
    logging.info("entry test")
    return "test"

async def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return await func.AsgiMiddleware(app).handle_async(req, context)

Those entries make it into my logs

Then in route1 I have

from fastapi import APIRouter
import logging

router=APIRouter(prefix="/api/route1")

@router.get("/test")
def test():
    logging.info("resp route test")
    return "route test"

The route itself works, that is to say if I navigate to 'api/route1/test' I get "resp route test" in my browser but then when I check the logs, I don't have the "route test" entry.

I tried adding logging.config.dictConfig({"disable_existing_loggers": False, "version": 1}) to the top of my route1.py file but it didn't help.

Answer 1

I just solved this problem and discovered a few quirks with logging on Azure Function Apps / Application Insights

• From what I can tell, the only logger that works on Azure is the root logger. That's logging.info() or logger = logging.getLogger(). print() statements do not work. Created loggers (customer_logger = logging.getLogger(__name__)) do not work.
• I added custom configuration to the root logger, and that broke it on Azure. (I'm not sure exactly which part broke it... my guess is some stream handling configuration. Also, logging.config.dictConfig(logging_config) threw an exception when running locally, using func host start.)
• By default, logs from created threads do not work.
  • @Dean this explains the behavior you were seeing, where logging.info('stuff happened in entrypoint') worked, but logging.info('stuff happened in route').
  • Background on how FastAPI handles requests: Asynchronous endpoints (async def function_name) run on the main thread, in an event loop. Synchronous endpoints (def function_name) each get their own thread. This way, if your API gets 5 requests at once, they don't block one another.
  • @Dean your synchronous endpoints wouldn't log because they were executing from created threads. Logging from async endpoints should work by default (per @Ikhtesam's answer above), but be careful, because if your async endpoints contain, e.g. synchronous database calls, they will block the main thread and you will lose all the concurrent goodness that FastAPI provides out-of-the-box.
  • Function Apps must co-opt the logging module in some way, because in order to log from a created thread, you have to tie that thread back to the parent thread's invocation_id (an Azure concept). There is an object that Azure passes down with each request, called the "context". The "context" contains an attribute thread_local_storage. thread_local_storage has an attribute called invocation_id and you need to set it from the created thread. Somehow, once that is set, the logs work.
  • There is a generic example here, which doesn't look too familiar if you're using FastAPI...

In order to fix this you need to first pass 1) the Azure context object and 2) the parent thread's invocation_id down to your endpoint, and then, from the created thread / endpoint, you need to set thread_local_storage.invocation_id to the parent's invocation_id.

1. Prepare the endpoint to receive the context

There is an implicit parameter for each endpoint called request. If you just define it as a parameter, you can use it in your endpoint and inspect it in a debugger.

from fastapi import Request

@app.get("/api/test")
def hello_world(request: Request):
    logging.info("hello world!") # remember print() doesn't work

Inside that request object, there is a dictionary called scope.

2a. AsgiFunctionApp / AsgiMiddleware

If you're using azure.functions.AsgiFunctionApp or azure.function.AsgiMiddleware (of a certain version, not exactly sure, but more recent than 1.8 I think), the context's thread_local_storage and the parent thread's invocation_id are actually included in the scope for you.

In that case, you would do the following in your synchronous endpoint (remember, no need to do this in an async endpoint)

from fastapi import Request

@app.get("/api/test")
def hello_world(request: Request):
    thread_local_storage = request.scope.get("azure_functions.thread_local_storage")
    if thread_local_storage is not None:
        parent_invocation_id = request.scope.get("azure_functions.invocation_id")
        thread_local_storage.invocation_id = parent_invocation_id
    logging.info("hello world!") # remember print() doesn't work

2b. Custom middleware

If you're using custom middleware, like this example here, you'll need to add the parent invocation_id and the context (or the thread_local_storage object) to the scope.

Something like this:

Then you would set it in your endpoint like so:

from fastapi import Request

@app.get("/api/test")
def hello_world(request: Request):
    context_obj = request.scope.get("azure_context", {}).get("context_obj")
    if context_obj and hasattr(context_obj, "thread_local_storage"):
        invocation_id = request.scope.get("azure_context", {}).get("invocation_id")
        if invocation_id:
            context_obj.thread_local_storage.invocation_id = request.scope.get("azure_context", {}
            ).get("invocation_id")
    logging.info("hello world!") # remember print() doesn't work

3. (Optional) Wrapper function

It's annoying to include that code at the beginning of all your functions, so you can create a wrapper function for it (h/t Eugene Nikolaiev)

from fastapi import Request
from functools import wraps

def logging_context_wrapper(func):
    """Passes invocation_id to the thread local context to enable proper Azure Functions logging.

    Can be applied to a sync handler which has request: Request parameter - to get the context from.
    """

    request_param = next(
        (param for (param, annotation) in func.__annotations__.items() if annotation is Request),
        None,
    )
    if not request_param:
        raise Exception("Function must have a request parameter of type Request.")

    @wraps(func)
    def wrapper(*args, **kwargs):
        request = kwargs[request_param]
        # if you were 2b from above, sub whatever code you used
        thread_local_storage = request.scope.get("azure_functions.thread_local_storage")
        if thread_local_storage is not None:
            thread_local_storage.invocation_id = request.scope.get("azure_functions.invocation_id")
        return func(*args, **kwargs)

    return wrapper



@app.get("/api/test")
@logging_context_wrapper
def hello_world(request: Request):
    logging.info("hello world!") # remember print() doesn't work

Summary

• Don't use print(), just logging.info(), logging.warning(), etc.
• Don't add configuration to logging, at least until you've got it working.
• Logging from synchronous endpoints (def foo, as opposed to async def foo) won't work, by default, since synchronous endpoints are run in separate threads.
• To log from synchronous endpoints, you need to set the context's thread_local_storage.invocation_id in the created thread / endpoint to the parent thread's invocation_id

Things I would for someone to explain

• How does Azure logging work and why don't print() statements or created loggers work?
• How is the thread pool managed on Function Apps? I can set an environment variable called PYTHON_THREADPOOL_THREAD_COUNT. Is that dictating how many threads FastAPI has available to it?