Skip to content

FastAPI Node

The FastAPI extra node can be used by adding the following to pyproject.toml
pyproject.toml
[tool.poetry.dependencies]
aineko = {version = "^0.2.7", extras=["fastapi"]}

API Reference

aineko.extras.fastapi.main.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 consumers and producers from your FastAPI app, import the consumers and producers variables from aineko.extras.fastapi. Use them as you would use self.consumers and self.producers in a regular node.

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

Example usage in pipeline.yml:

pipeline.yml
pipeline:
  nodes:
    fastapi:
      class: aineko.extras.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.extras.fastapi import consumers, producers

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

Source code in aineko/extras/fastapi/main.py 
 72
 73
 74
 75
 76
 77
 78
 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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class FastAPI(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 consumers and producers from your FastAPI app, import the
    `consumers` and `producers` variables from `aineko.extras.fastapi`. Use
    them as you would use `self.consumers` and `self.producers` in a regular
    node.

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

    Example usage in pipeline.yml:
    ```yaml title="pipeline.yml"
    pipeline:
      nodes:
        fastapi:
          class: aineko.extras.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](https://fastapi.tiangolo.com/){:target="\_blank"}
    on how to create a FastAPI app.

    Example usage in FastAPI app:
    ```python title="fastapi.py"
    from aineko.extras.fastapi import consumers, producers

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

    def _pre_loop_hook(self, params: Optional[dict] = None) -> None:
        """Initialize node state. Set env variables for Fast API app."""
        for key, value in self.consumers.items():
            consumers[key] = value
        for key, value in self.producers.items():
            producers[key] = value

    def _execute(self, params: dict) -> None:
        """Start the API server."""
        config = uvicorn.Config(
            app=params.get("app"),  # type: ignore
            port=params.get("port", 8000),
            log_level=params.get("log_level", "info"),
            host="0.0.0.0",
        )
        server = uvicorn.Server(config)
        server.run()