Skip to content

FastAPI Classes

Custom FastAPI router classes that extend the standard FastAPI functionality for DiracX-specific features.

These classes provide enhanced router behavior such as automatic access policy enforcement, extension support, and DiracX-specific metadata.

fastapi_classes

Attributes

T = TypeVar('T') module-attribute

Classes

DiracFastAPI

Bases: FastAPI

Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class DiracFastAPI(FastAPI):
    def __init__(self):
        @contextlib.asynccontextmanager
        async def lifespan(app: DiracFastAPI):
            async with contextlib.AsyncExitStack() as stack:
                await asyncio.gather(
                    *(stack.enter_async_context(f()) for f in app.lifetime_functions)
                )
                yield

        self.lifetime_functions = []
        super().__init__(
            swagger_ui_init_oauth={
                "clientId": "myDIRACClientID",
                "scopes": "property:NormalUser",
                "usePkceWithAuthorizationCodeGrant": True,
            },
            generate_unique_id_function=lambda route: f"{route.tags[0]}_{route.name}",
            title="Dirac",
            lifespan=lifespan,
            openapi_url="/api/openapi.json",
            docs_url="/api/docs",
            swagger_ui_oauth2_redirect_url="/api/docs/oauth2-redirect",
        )
        # FIXME: when autorest will support 3.1.0
        # From 0.99.0, FastAPI is using openapi 3.1.0 by default
        # This version is not supported by autorest yet
        self.openapi_version = "3.0.2"

    def openapi(self, *args, **kwargs):
        if not self.openapi_schema:
            super().openapi(*args, **kwargs)
            _downgrade_openapi_schema(self.openapi_schema)

            # Remove 422 responses as we don't want autorest to use it
            for _, method_item in self.openapi_schema.get("paths").items():
                for _, param in method_item.items():
                    responses = param.get("responses")
                    if "422" in responses:
                        del responses["422"]

        return self.openapi_schema
Attributes
lifetime_functions = [] instance-attribute
openapi_version = '3.0.2' instance-attribute
Functions
__init__()
Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(self):
    @contextlib.asynccontextmanager
    async def lifespan(app: DiracFastAPI):
        async with contextlib.AsyncExitStack() as stack:
            await asyncio.gather(
                *(stack.enter_async_context(f()) for f in app.lifetime_functions)
            )
            yield

    self.lifetime_functions = []
    super().__init__(
        swagger_ui_init_oauth={
            "clientId": "myDIRACClientID",
            "scopes": "property:NormalUser",
            "usePkceWithAuthorizationCodeGrant": True,
        },
        generate_unique_id_function=lambda route: f"{route.tags[0]}_{route.name}",
        title="Dirac",
        lifespan=lifespan,
        openapi_url="/api/openapi.json",
        docs_url="/api/docs",
        swagger_ui_oauth2_redirect_url="/api/docs/oauth2-redirect",
    )
    # FIXME: when autorest will support 3.1.0
    # From 0.99.0, FastAPI is using openapi 3.1.0 by default
    # This version is not supported by autorest yet
    self.openapi_version = "3.0.2"
openapi(*args, **kwargs)
Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def openapi(self, *args, **kwargs):
    if not self.openapi_schema:
        super().openapi(*args, **kwargs)
        _downgrade_openapi_schema(self.openapi_schema)

        # Remove 422 responses as we don't want autorest to use it
        for _, method_item in self.openapi_schema.get("paths").items():
            for _, param in method_item.items():
                responses = param.get("responses")
                if "422" in responses:
                    del responses["422"]

    return self.openapi_schema

DiracxRouter

Bases: APIRouter

Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class DiracxRouter(APIRouter):
    def __init__(
        self,
        *,
        dependencies=None,
        require_auth: bool = True,
        include_in_schema: bool = True,
        path_root: str = "/api",
    ):
        super().__init__(dependencies=dependencies, include_in_schema=include_in_schema)
        self.diracx_require_auth = require_auth
        self.diracx_path_root = path_root

    ####
    # These 2 methods are needed to overwrite routes
    # https://github.com/tiangolo/fastapi/discussions/8489

    def add_api_route(self, path: str, endpoint: Callable[..., Any], **kwargs):
        route_index = self._get_route_index_by_path_and_methods(
            path, set(kwargs.get("methods", []))
        )
        if route_index >= 0:
            self.routes.pop(route_index)

        return super().add_api_route(path, endpoint, **kwargs)

    def _get_route_index_by_path_and_methods(self, path: str, methods: set[str]) -> int:
        routes = cast(list[Route], self.routes)
        for index, route in enumerate(routes):
            if route.path == path and methods == route.methods:
                return index
        return -1
Attributes
diracx_require_auth = require_auth instance-attribute
diracx_path_root = path_root instance-attribute
Functions
__init__(*, dependencies=None, require_auth=True, include_in_schema=True, path_root='/api')
Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    *,
    dependencies=None,
    require_auth: bool = True,
    include_in_schema: bool = True,
    path_root: str = "/api",
):
    super().__init__(dependencies=dependencies, include_in_schema=include_in_schema)
    self.diracx_require_auth = require_auth
    self.diracx_path_root = path_root
add_api_route(path, endpoint, **kwargs)
Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
 96
 97
 98
 99
100
101
102
103
def add_api_route(self, path: str, endpoint: Callable[..., Any], **kwargs):
    route_index = self._get_route_index_by_path_and_methods(
        path, set(kwargs.get("methods", []))
    )
    if route_index >= 0:
        self.routes.pop(route_index)

    return super().add_api_route(path, endpoint, **kwargs)
_get_route_index_by_path_and_methods(path, methods)
Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
105
106
107
108
109
110
def _get_route_index_by_path_and_methods(self, path: str, methods: set[str]) -> int:
    routes = cast(list[Route], self.routes)
    for index, route in enumerate(routes):
        if route.path == path and methods == route.methods:
            return index
    return -1

Functions

_downgrade_openapi_schema(data)

Modify an openapi schema in-place to be compatible with AutoRest.

Source code in diracx-routers/src/diracx/routers/fastapi_classes.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def _downgrade_openapi_schema(data):
    """Modify an openapi schema in-place to be compatible with AutoRest."""
    if isinstance(data, dict):
        for k, v in list(data.items()):
            if k == "anyOf":
                if {"type": "null"} in v:
                    v.pop(v.index({"type": "null"}))
                    data["nullable"] = True
                    if len(v) == 1:
                        data |= v[0]
            elif k == "const":
                data.pop(k)
            # https://github.com/fastapi/fastapi/discussions/12984
            elif k == "propertyNames":
                data.pop(k)

            _downgrade_openapi_schema(v)
    if isinstance(data, list):
        for v in data:
            _downgrade_openapi_schema(v)