FastAPI Node

Installation

The FastAPI plugin can be installed using poetry with the following command poetry add aineko-plugins-nodes-fastapi-server.

API Reference

aineko_plugins.nodes.fastapi_server.FastAPI

Bases: AbstractNode

Node for creating a FastAPI app with a gunicorn server.

node_params should contain the following keys:

app: path to FastAPI app
port (optional): port to run the server on. Defaults to 8000.
log_level (optional): log level to log messages from the uvicorn server.
    Defaults to "info".

To access the inputs and outputs from your FastAPI app, import the inputs and outputs variables from aineko_plugins.nodes.fastapi_server. Use them as you would use self.inputs and self.outputs in a regular node.

We recommend no more than 1 FastAPI node per pipeline since the Inputs and Outputs objects are namespaced at the pipeline level.

Example usage in pipeline.yml:

pipeline.yml

pipeline:
  nodes:
    fastapi:
      class: aineko_plugins.nodes.fastapi_server.FastAPI
      inputs:
        - test_sequence
      node_params:
        app: my_awesome_pipeline.fastapi:app
        port: 8000

where the app points to a FastAPI app. See FastAPI documentation on how to create a FastAPI app.

Example usage in FastAPI app:

fastapi.py

from aineko_plugins.nodes.fastapi_server import inputs, outputs

@app.get("/query")
async def query():
    msg = inputs["test_sequence"].next()
    return msg

Health Check Endpoint

A common use case with an API server is to check its operational health status. For Aineko specifically, it is useful to know if the pipeline and API node are operational as well. Aineko comes with a light weight helper function as follows:

aineko_plugins.nodes.fastapi_server.health.read_health async

read_health() -> dict

Return the health status of the Aineko FastAPI server.

Health router can be imported by an Aineko FastAPI server and added to the app as a router.

The health check endpoint is available via a GET request to the /health route and returns a 200 response if the server is active.

For example, if the FastAPI app is running locally on http://localhost:8000, the health check endpoint can be accessed via a GET request query to http://localhost:8000/health.

Example usage in FastAPI app

from fastapi import FastAPI
from aineko_plugins.nodes.fastapi_server import health_router

app = FastAPI()
app.include_router(health_router)

Source code in nodes/aineko-plugins-nodes-fastapi-server/aineko_plugins/nodes/fastapi_server/health.py

@health_router.get("/health", status_code=200)
async def read_health() -> dict:
    """Return the health status of the Aineko FastAPI server.

    Health router can be imported by an Aineko FastAPI server and added
    to the app as a router.

    The health check endpoint is available via a GET request to the `/health`
    route and returns a 200 response if the server is active.

    For example, if the FastAPI app is running locally on
    `http://localhost:8000`, the health check endpoint can be accessed via
    a GET request query to `http://localhost:8000/health`.

    Example usage in FastAPI app:
        ```python hl_lines="2 5"
        from fastapi import FastAPI
        from aineko_plugins.nodes.fastapi_server import health_router

        app = FastAPI()
        app.include_router(health_router)
        ```
    """
    return {"status": "healthy"}

Authentication Considerations

By default, the /health endpoint does not contain authentication.

One method to add authentication is to use security middleware at the app level for FastAPI. This will handle adding authentication to the health endpoint without having to add custom dependencies.

For more information about security with FastAPI, check out the documentation here.