FastAPI support

[brikelly]

appmap-python will record AppMaps of FastAPI code via test case recording.

Enhance appmap-python to support requests recording and remote recording for FastAPI: https://fastapi.tiangolo.com/

Ensure that the output includes HTTP server requests and responses.

[kgilpin]

Per @apotterri FastAPI is built on Starlette, HTTP framework that implements ASGI. The 'A' in ASGI stands for "Asynchronous", and there's no requirement that there be a single thread that processes an entire request.

Adding support for it is almost certainly going to be more complicated than just hooking a couple of framework methods.

[mavwolverine]

@apotterri
took some inspiration from sentry_sdk

import asyncio
from functools import wraps
import threading
import time

import fastapi

from _appmap.env import Env
from _appmap.metadata import Metadata

logger = Env.current.getLogger(__name__)


try:
    from gevent import get_hub as get_gevent_hub  # type: ignore
    from gevent.monkey import get_original, is_module_patched  # type: ignore
    from gevent.threadpool import ThreadPool  # type: ignore

    thread_sleep = get_original("time", "sleep")
except ImportError:

    def get_gevent_hub():
        return None

    thread_sleep = time.sleep

    def is_module_patched(*args, **kwargs):
        # unable to import from gevent means no modules have been patched
        return False

    ThreadPool = None


def is_gevent():
    return is_module_patched("threading") or is_module_patched("_thread")


def get_current_thread_id(thread=None):
    """
    Try to get the id of the current thread, with various fall backs.
    """

    # if a thread is specified, that takes priority
    if thread is not None:
        try:
            thread_id = thread.ident
            if thread_id is not None:
                return thread_id
        except AttributeError:
            pass

    # if the app is using gevent, we should look at the gevent hub first
    # as the id there differs from what the threading module reports
    if is_gevent():
        gevent_hub = get_gevent_hub()
        if gevent_hub is not None:
            try:
                # this is undocumented, so wrap it in try except to be safe
                return gevent_hub.thread_ident
            except AttributeError:
                pass

    # use the current thread's id if possible
    try:
        current_thread_id = threading.current_thread().ident
        if current_thread_id is not None:
            return current_thread_id
    except AttributeError:
        pass

    # if we can't get the current thread id, fall back to the main thread id
    try:
        main_thread_id = threading.main_thread().ident
        if main_thread_id is not None:
            return main_thread_id
    except AttributeError:
        pass

    # we've tried everything, time to give up
    return None


def patch_get_request_handler():
    old_get_request_handler = fastapi.routing.get_request_handler

    def _appmap_get_request_handler(*args, **kwargs):
        dependant = kwargs.get("dependant")
        if (
            dependant
            and dependant.call is not None
            and not asyncio.iscoroutinefunction(dependant.call)
        ):
            old_call = dependant.call

            @wraps(old_call)
            def _appmap_call(*args, **kwargs):
                print('-' * 50)
                print('_appmap_call')

                print('get_current_thread_id', get_current_thread_id())
                return old_call(*args, **kwargs)

            dependant.call = _appmap_call

        old_app = old_get_request_handler(*args, **kwargs)

        async def _appmap_app(*args, **kwargs):
            Metadata.add_framework("fastapi", fastapi.__version__)

            print('-' * 50)
            print('_appmap_app')

            request = args[0]
            print(request)

            response = await old_app(*args, **kwargs)
            print(response)

            return response

        return _appmap_app

    fastapi.routing.get_request_handler = _appmap_get_request_handler


if Env.current.enabled:
    patch_get_request_handler()

Please let me know if you think this might work.

I haven't had a chance to go through MiddlewareInserter and AppmapMiddleware, so don't really understand how to use it yet here. fastapi.routing.get_request_handler gets called on every request.

[mavwolverine]

Another thing to note is, appmap still supports Python 3.7.2, but FastAPI requires >=3.8

[mavwolverine]

@apotterri @kgilpin FastAPI supports middlewares, but injecting them like flask and django doesn't seem to be possible, SO FAR.

Since FastAPI is run using uvicorn, there is no common fixed entry point inside Starlette or FastAPI. Need to explore if we can somehow inject directly inside uvicorn.

Else, we can create middleware and would need the dev to do something like

app = FastAPI()
app.add_middleware(AppmapFastAPIMiddleware)

https://semaphoreci.com/blog/custom-middleware-fastapi
https://stackoverflow.com/questions/71525132/how-to-write-a-custom-fastapi-middleware-class

[apotterri]

@virajkanwade welcome back, and thanks a lot for digging into this!

I think it would be better to implement an ASGI middleware to manage request recording, rather than tying ourselves to FastAPI specifically. This would mean that we'd work with any ASGI server, and so also support all the frameworks that run on them. It might also mean that we won't be limited by FastAPI's 3.8 requirement (though that will depend on how we test the new middleware, I guess).

Implementing the middleware itself should be pretty straightforward, as it will look a lot like AppmapMiddleware (and they may even be able to share some code).

Maybe somewhat trickier will be changing the way the current Recorder is managed. Currently, it's stored in a thread-local variable:

appmap-python/_appmap/utils.py

Line 64 in 2c2025b

_appmap_tls = ThreadLocalDict()

. It seems like changing the implementation to use a ContextVar instead of a threading.local will be sufficient, but I'm not 100% sure.

We should definitely see if we can hook something in uvicorn (and any other support we want to support) to inject the new middleware. The less the user has to do to get AppMaps, the better.

[kgilpin]

@virajkanwade thanks for diving into this! FastAPI would be an awesome enhancement.