FastRAMQPI 1.5.3

Rammearkitektur AMQP framework (FastAPI + RAMQP)

FastRAMQPI

FastRAMQPI is an opinionated library for FastAPI and RAMQP.

It is implemented as a thin wrapper around FastAPI and RAMQP. It is very MO specific.

Usage

from pydantic import BaseSettings
from fastramqpi import FastRAMQPI
from fastramqpi import FastRAMQPISettings


class Settings(BaseSettings):
    class Config:
        frozen = True
        env_nested_delimiter = "__"

    fastramqpi: FastRAMQPISettings = Field(
        default_factory=FastRAMQPISettings,
        description="FastRAMQPI settings"
    )

    # All your program settings hereunder...


fastapi_router = APIRouter()

@fastapi_router.post("/trigger/all")
async def update_all(request: Request) -> dict[str, str]:
    context: dict[str, Any] = request.app.state.context
    graphql_session = context["grapqh_session"]
    program_settings = context["user_context"]["settings"]
    ...
    return {"status": "OK"}


amqp_router = MORouter()

@amqp_router.register("*.*.*")
async def listen_to_all(context: dict, payload: PayloadType) -> None:
    graphql_session = context["grapqh_session"]
    program_settings = context["user_context"]["settings"]
    ...


def create_fastramqpi(**kwargs: Any) -> FastRAMQPI:
    settings = Settings(**kwargs)
    fastramqpi = FastRAMQPI(
        application_name="orggatekeeper", settings=settings.fastramqpi
    )
    fastramqpi.add_context(settings=settings)

    # Add our AMQP router(s)
    amqpsystem = fastramqpi.get_amqpsystem()
    amqpsystem.router.registry.update(amqp_router.registry)

    # Add our FastAPI router(s)
    app = fastramqpi.get_app()
    app.include_router(fastapi_router)

    return fastramqpi


def create_app(**kwargs: Any) -> FastAPI:
    fastramqpi = create_fastramqpi(**kwargs)
    return fastramqpi.get_app()

Metrics

FastRAMQPI Metrics are exported via prometheus/client_python on the FastAPI's /metrics.

Autogenerated GraphQL Client

FastRAMQPI exposes an authenticated httpx client through the dependency injection system. While it is possible to call the OS2mo API directly through it, the recommended approach is to define a properly-typed GraphQL client in the integration and configure it to make calls through the authenticated client. Instead of manually implementing such client, we strongly recommend to use the Ariadne Code Generator, which generates an integration-specific client based on the general OS2mo GraphQL schema and the exact queries and mutations the integration requires.

To integrate such client, first add and configure the codegen:

# pyproject.toml

[tool.poetry.dependencies]
ariadne-codegen = {extras = ["subscriptions"], version = "^0.7.1"}

[tool.ariadne-codegen]
# Ideally, the GraphQL client is generated as part of the build process and
# never committed to git. Unfortunately, most of our tools and CI analyses the
# project directly as it is in Git. In the future - when the CI templates
# operate on the built container image - only the definition of the schema and
# queries should be checked in.
#
# The default package name is `graphql_client`. Make it more obvious that the
# files are not to be modified manually.
target_package_name = "autogenerated_graphql_client"
target_package_path = "my_integration/"
client_name = "GraphQLClient"
schema_path = "schema.graphql"  # curl -O http://localhost:5000/graphql/v8/schema.graphql
queries_path = "queries.graphql"
plugins = [
    # Return values directly when only a single top field is requested
    "ariadne_codegen.contrib.shorter_results.ShorterResultsPlugin",
]
[tool.ariadne-codegen.scalars.DateTime]
type = "datetime.datetime"
[tool.ariadne-codegen.scalars.UUID]
type = "uuid.UUID"

Grab OS2mo's GraphQL schema:

curl -O http://localhost:5000/graphql/v8/schema.graphql

Define your queries:

# queries.graphql

# SPDX-FileCopyrightText: Magenta ApS <https://magenta.dk>
# SPDX-License-Identifier: MPL-2.0

query Version {
  version {
    mo_version
    mo_hash
  }
}

Generate the client - you may have to activate some virtual environment:

ariadne-codegen

The client class is passed to FastRAMQPI on startup. This will ensure it is automatically opened and closed and configured with authentication:

# app.py
from autogenerated_graphql_client import GraphQLClient


def create_app(**kwargs: Any) -> FastAPI:
    fastramqpi = FastRAMQPI(..., graphql_client_cls=GraphQLClient)
    ...

The FastRAMQPI framework cannot define the annotated type for the GraphQL client since its methods depend on the specific queries required by the integration. Therefore, each implementing integration needs to define their own:

# depends.py
from typing import Annotated

from fastapi import Depends
from ramqp.depends import from_context

from my_integration.autogenerated_graphql_client import GraphQLClient as _GraphQLClient

GraphQLClient = Annotated[_GraphQLClient, Depends(from_context("graphql_client"))]

Finally, we can define our AMQP handler to use the GraphQL client:

# events.py
from . import depends


@router.register("*")
async def handler(mo: depends.GraphQLClient) -> None:
    version = await mo.version()
    print(version)

To get REUSE working, you might consider adding the following to .reuse/dep5:

Files: my_integration/autogenerated_graphql_client/*
Copyright: Magenta ApS <https://magenta.dk>
License: MPL-2.0

Development

Prerequisites

• Poetry

Getting Started

1. Clone the repository:

git clone [e-mail]:rammearkitektur/FastRAMQPI.git

2. Install all dependencies:

poetry install

3. Set up pre-commit:

poetry run pre-commit install

Running the tests

You use poetry and pytest to run the tests:

poetry run pytest

You can also run specific files

poetry run pytest tests/<test_folder>/<test_file.py>

and even use filtering with -k

poetry run pytest -k "Manager"

You can use the flags -vx where v prints the test & x makes the test stop if any tests fails (Verbose, X-fail)

Authors

Magenta ApS https://magenta.dk

License

This project uses: MPL-2.0

This project uses REUSE for licensing. All licenses can be found in the LICENSES folder of the project.