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.

[virajkanwade]

@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.

[virajkanwade]

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

[virajkanwade]

@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.

[virajkanwade]

@apotterri uvicorn injection

from typing import List, Optional, Tuple, Union, cast

import uvicorn
from uvicorn._types import (
    ASGI3Application,
    ASGIReceiveCallable,
    ASGISendCallable,
    ASGISendEvent,
    Scope,
)
from uvicorn.config import Config

from _appmap import wrapt
from _appmap.env import Env
from _appmap.web_framework import AppmapMiddleware, MiddlewareInserter


class AppmapUvicornMiddleware:
    def __init__(
        self,
        app: "ASGI3Application",
    ) -> None:
        self.app = app

    async def __call__(
        self, scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
    ) -> None:
        # scope["type"]
        #   "http": for HTTP requests.
        #   "websocket": for WebSocket connections.
        #   "lifespan": for ASGI lifespan messages.

        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client_addr: Optional[Tuple[str, int]] = scope.get("client")
        client_host = client_addr[0] if client_addr else None

        print("client_host", client_host)
        print("scheme", scope["scheme"])
        print("method", scope["method"])
        print("path", scope["path"])
        print("query_string", scope["query_string"])
        print("headers", scope["headers"])

        print(scope)

        async def send_wrapper(message: ASGISendEvent) -> None:
            print(message)

            await send(message)

        return await self.app(scope, receive, send_wrapper)


def install_extension(wrapped, self_obj, args, kwargs):
    wrapped(*args, **kwargs)
    app = self_obj.loaded_app
    if app:
        app = AppmapUvicornMiddleware(app)

        self_obj.loaded_app = app


if Env.current.enabled:
    Config.load = wrapt.wrap_function_wrapper("uvicorn.config", "Config.load", install_extension)

[apotterri]

Nice! Thank for this.

[virajkanwade]

@apotterri hypercorn injection

from typing import Callable

from hypercorn import utils
from hypercorn.typing import (
    ASGIFramework,
    ASGIReceiveCallable,
    ASGISendCallable,
    ASGISendEvent,
    Scope,
)

from _appmap import wrapt
from _appmap.env import Env


class AppmapHypercornMiddleware:
    def __init__(
        self,
        app: "ASGIFramework",
    ) -> None:
        self.app = app

    async def __call__(
        self,
        scope: "Scope",
        receive: "ASGIReceiveCallable",
        send: "ASGISendCallable",
        sync_spawn: Callable,
        call_soon: Callable,
    ) -> None:
        # scope["type"]
        #   "http": for HTTP requests.
        #   "websocket": for WebSocket connections.
        #   "lifespan": for ASGI lifespan messages.

        if scope["type"] != "http":
            await self.app(scope, receive, send, sync_spawn, call_soon)
            return

        client_addr = scope.get("client")
        client_host = client_addr[0] if client_addr else None

        print("client_host", client_host)
        print("scheme", scope["scheme"])
        print("method", scope["method"])
        print("path", scope["path"])
        print("query_string", scope["query_string"])
        print("headers", scope["headers"])

        print(scope)

        async def send_wrapper(message: ASGISendEvent) -> None:
            print(message)

            await send(message)

        return await self.app(scope, receive, send_wrapper, sync_spawn, call_soon)


def install_extension(wrapped, _, args, kwargs):
    app = wrapped(*args, **kwargs)

    if app:
        app = AppmapHypercornMiddleware(app)

    return app


if Env.current.enabled:
    utils.load_application = wrapt.wrap_function_wrapper(
        "hypercorn.utils", "load_application", install_extension
    )

This is slightly different from uvicorn since the middleware.__call__ expects 2 additional params, even though they are not used anywhere even in hypercorn. I have raised a request pgjones/hypercorn#188. If hypercorn implements it, the same asgi middleware could be used here too.

Works with both
hypercorn main:app
and
hypercorn main:app --worker-class trio

[virajkanwade]

@apotterri there seems to be weird issue with daphne and appmap-python.

The startup log is not printed, but the daphne server works.

Comment out this line https://github.com/getappmap/appmap-python/blob/master/_appmap/env.py#L168 and it starts printing the startup logs.

2024-02-09 16:15:24,452 INFO     Starting server at tcp:port=8000:interface=127.0.0.1
2024-02-09 16:15:24,452 INFO     HTTP/2 support enabled
2024-02-09 16:15:24,452 INFO     Configuring endpoint tcp:port=8000:interface=127.0.0.1
2024-02-09 16:15:24,453 INFO     Listening on TCP address 127.0.0.1:8000
^C2024-02-09 16:15:28,484 INFO     Killed 0 pending application instances

[virajkanwade]

@apotterri daphne injection

from asgiref.typing import (
    ASGI3Application,
    ASGIReceiveCallable,
    ASGISendCallable,
    ASGISendEvent,
    Scope,
)
from daphne.server import Server

from _appmap import wrapt
from _appmap.env import Env


class AppmapDaphneMiddleware:
    def __init__(
        self,
        app: "ASGI3Application",
    ) -> None:
        self.app = app

    async def __call__(
        self, scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
    ) -> None:
        # scope["type"]
        #   "http": for HTTP requests.
        #   "websocket": for WebSocket connections.
        #   "lifespan": for ASGI lifespan messages.

        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client_addr = scope.get("client")
        client_host = client_addr[0] if client_addr else None

        print("client_host", client_host)
        print("scheme", scope["scheme"])
        print("method", scope["method"])
        print("path", scope["path"])
        print("query_string", scope["query_string"])
        print("headers", scope["headers"])

        print(scope)

        async def send_wrapper(message: ASGISendEvent) -> None:
            print(message)

            await send(message)

        return await self.app(scope, receive, send_wrapper)


def install_extension(wrapped, self_obj, args, kwargs):
    app = self_obj.application
    if app:
        self_obj.application = AppmapDaphneMiddleware(app)

    wrapped(*args, **kwargs)


if Env.current.enabled:
    Server.run = wrapt.wrap_function_wrapper("daphne.server", "Server.run", install_extension)

Almost similar to uvicorn. Only differences are

• uvicorn has self_obj.loaded_app and daphne has self_obj.application
• uvicorn needs to call wrapped function before and daphne needs to call it later

[apotterri]

@virajkanwade Thank you so much for the work on figuring out how to inject into the various servers. It's hugely helpful.

I should have time today to get started the changes necessary to implement an ASGI middleware to manage recording.

Thanks again.

[EwertonDCSilv]

Temos alguma novidade sobre o tema ? Algo simplificado ? Ou a injeção é o único caminho ?

[apotterri]

@EwertonDCSilv thanks for your interest in AppMap!

I'm continuing to work on FastAPI support. There's a PR open for it here: #282, which I'm expecting to update later today or tomorrow.

I'm not sure when it will be finished, though I expect it should be soon.

[EwertonDCSilv]

@apotterri alguma novidade ?

[apotterri]

@virajkanwade @EwertonDCSilv FastAPI integration is available in v1.20.0 of the Python agent, now available on PyPI: https://pypi.org/project/appmap/1.20.0/

[virajkanwade]

@apotterri for some reason, test_fastapi is failing for me

============================================================================================== short test summary info ===============================================================================================
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_path_normalization[/post/123-/post/{post_id}] - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_http_capture_post - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_get - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_get_arr - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_put - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_post_with_query - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_path_normalization[/123/posts/test_user-/{org}/posts/{username}] - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_path_normalization[/user/test_user-/user/{username}] - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_http_capture - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_message_path_segments - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_path_normalization[/post/test_user/123/summary-/post/{username}/{post_id}/summary] - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRequestCapture::test_post - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::test_framework_metadata - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRemoteRecording::test_can_record - AttributeError: '_CachedRequest' object has no attribute 'path'
FAILED _appmap/test/test_fastapi.py::TestRecordRequests::test_remote_disabled_in_prod - assert 500 == 404
FAILED _appmap/test/test_fastapi.py::TestRecordRequests::test_record_requests_with_remote - assert 500 == 200
FAILED _appmap/test/test_fastapi.py::TestRecordRequests::test_record_requests_without_remote - assert 500 == 200
==================================================================================== 17 failed, 188 passed, 22 warnings in 40.15s ====================================================================================

[apotterri]

Not sure why that would be, the CI builds are all green: https://app.travis-ci.com/github/getappmap/appmap-python/builds/269484671

Do the tests also fail if you run them using tox, e.g. tox --recreate -e py312-web?

[virajkanwade]

@apotterri seems like it was an env issue. After using the --recreate it is now passing. (altough there are 19 warnings)

[virajkanwade]

The warnings seem related to
encode/starlette#2524
fastapi/fastapi#5574

[EwertonDCSilv]

@virajkanwade @EwertonDCsilva integração FastAPI está disponível na v1.20.0 do agente Python, agora disponível no PyPI: https://pypi.org/project/appmap/1.20.0/

Como ficou a configuração para as chamadas http do Fastapi ?

[EwertonDCSilv]

@virajkanwade @EwertonDCsilva integração FastAPI está disponível na v1.20.0 do agente Python, agora disponível no PyPI: https://pypi.org/project/appmap/1.20.0/

Como ficou a configuração para as chamadas http do Fastapi ?

@apotterri ficaria muito grato com um exemplo básico !

[apotterri]

@EwertonDCSilv The documentation for the Python agent has been updated to include a description of using it with a FastAPI application: https://appmap.io/docs/reference/appmap-python#web-framework-support .