FastAPI
=======

To test the scripts provided in this page, you can use the following methods:

Using python venv:

    .. code-block:: bash

        python3 -m venv venv-fastapi
        source venv-fastapi/bin/activate
        pip install "fastapi[standard]" aiofiles


    Save the code to test to the file `simple_api.py` and run the app using the command:

    .. code-block:: bash

        python3 -m fastapi dev simple_api.py
        # OR
        fastapi dev simple_api.py

    .. note::
        The default port used is :code:`8000`, you can change it using the argument :code:`--port` (for example: :code:`fastapi dev simple_api.py --port 5000`)


You can also use docker to run the APP:

    #. Create the :code:`Dockerfile`:

        .. code-block:: dockerfile

            FROM python:3.12-slim
            WORKDIR /app
            COPY simple_api.py /app
            RUN pip install "fastapi[standard]" aiofiles
            CMD ["python3", "-m", "fastapi", "dev", "simple_api.py"]

    #. Save the code to test to the file `simple_api.py`

    #. Run the :code:`Dockerfile`:

        .. code-block:: bash

            docker build -t fastapi-app . && docker run --rm --network host -it fastapi-app


Simple API
##########

Receiving data from the client
******************************

.. note::
    As explained in the `official documentation <https://fastapi.tiangolo.com/tutorial/request-forms/>`_ , if you're using :code:`Form` to receive form data in the request, you need to make sure you installed the python package :code:`python-multipart` (package automatically installed if you installed FastAPI using :code:`fastapi[standard]`)

.. note::
    To save a file received in a form, we use the asynchronous way explained in this `stackoverflow question <https://stackoverflow.com/questions/63580229/how-to-save-uploadfile-in-fastapi>`_ , for that we need to install an additional python package :code:`aiofiles`


.. code-block:: python

    from fastapi import FastAPI, Request, Header, Cookie, Form, UploadFile
    from pydantic import BaseModel
    from typing import Annotated
    import aiofiles


    app = FastAPI()

    # Query parameters
    @app.get("/query")
    def query_paramters(p1: str, p2: str | None = None):
        print(f"p1: {p1}")
        print(f"p2 (optional): {p2}")
        return ""

    # Path parameters
    @app.post("/path/{username}/{post_id}")
    def path_paramters(username: str, post_id: int):
        print(f'username: {username}, post_id: {post_id}')
        return ""

    # Request body parameters
    @app.post("/body_raw")
    async def body_raw_parameters(request: Request):
        raw_data = await request.body()
        print(f'raw_data: {raw_data}')
        return ""

    class Person(BaseModel):
        name: str
        age: float

    @app.post("/body_json")
    async def body_json_parameters(person: Person):
        print(person)
        return ""

    # HTTP headers parameters
    # List of headers: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers
    @app.get("/header")
    async def header_parameters(user_agent: Annotated[str | None, Header()] = None):
        print(user_agent)
        return ""

    # Cookies parameters
    @app.get("/cookies")
    async def cookies_parameters(username: Annotated[str | None, Cookie()] = None):
        print(username)
        return ""

    # Form Data (URL-encoded or Multipart)
    @app.post("/form")
    async def form_parameters(
        name: Annotated[str, Form()],
        email: Annotated[str, Form()],
        file_in: UploadFile | None = None,
        content_type: Annotated[str | None, Header()] = None):
        print(f"Content-Type: {content_type}")
        if 'multipart/form-data' in content_type:
            print(f"Content of 'file_in': {await file_in.read()}")
            await file_in.seek(0)
            async with aiofiles.open('uploaded_file.txt', 'wb') as out_file:
                content = await file_in.read()
                await out_file.write(content)
        print(f"name: {name}, email: {email}")
        return ""


You can test the routes using the :code:`curl` commands below:

.. code-block:: bash

    curl -X GET "http://localhost:8000/query?p1=value123"
    curl -X POST "http://localhost:8000/path/nicoh/101"
    curl -X POST "http://localhost:8000/body_raw" -d "This is raw data"
    curl -X POST "http://localhost:8000/body_json" -H "Content-Type: application/json" -d '{"name": "Hania <3", "age": 30}'
    curl -X GET "http://localhost:8000/header" -H "User-Agent: CustomUserAgent/1.0"
    curl -X GET "http://localhost:8000/cookies" --cookie "username=nicoh"
    curl -X POST "http://localhost:8000/form" -d "name=Nicolas&email=nicolas@example.com"
    echo "Coucou" > file_test.txt
    curl -X POST "http://localhost:8000/form" -F "name=Nicolas" -F "email=nicolas@example.com" -F "file_in=@file_test.txt"


Sending data to the client
**************************

.. code-block:: python

    from fastapi import FastAPI, status, Response, HTTPException
    from pydantic import BaseModel
    from fastapi.responses import HTMLResponse, PlainTextResponse, FileResponse, StreamingResponse
    from typing import Any
    from markupsafe import escape

    app = FastAPI()

    # HTML response
    @app.get('/html/{name}', response_class=HTMLResponse)
    async def html_response(name: str):
        # escape: To protect from injection attacks
        return f"<h1>Hello {escape(name)}</h1>"

    # TEXT response
    @app.get('/text', response_class=PlainTextResponse, status_code=status.HTTP_200_OK)
    async def txt_response():
        return "This is the content"

    # Explicit status code response
    tasks = {"foo": "Listen to the Bar Fighters"}
    @app.put("/get-or-create-task/{task_id}", status_code=200)
    async def get_or_create_task(task_id: str, response: Response):
        if task_id not in tasks:
            if task_id == "error":
                raise HTTPException(status_code=404, detail="Item not found")
            tasks[task_id] = "This didn't exist before"
            response.status_code = status.HTTP_201_CREATED
        return tasks[task_id]

    # json response
    @app.get('/json_any')
    async def json_response() -> Any:
        return {"message": "Hello, World!", "status": "success"}

    class Person(BaseModel):
        name: str
        age: int

    @app.get('/json_model')
    async def json_response() -> Person:
        return {"name": "Hania <3", "age": 33}

    # File
    @app.get('/download')
    async def download_file():
        filepath = "file_to_send.txt"
        with open(filepath, "w") as f:
            f.write("this is the file content")
        return FileResponse(filepath, filename="custom_filename.txt")

    @app.get('/download_2')
    async def download_file():
        filepath = "file_to_send.txt"
        with open(filepath, "w") as f:
            f.write("this is the file content 2")
        def iterfile():
            with open(filepath, mode="rb") as f:
                yield from f
        return StreamingResponse(iterfile(), media_type="text/plain")


You can test the routes using the :code:`curl` commands below:

.. code-block:: bash

    curl -X GET "http://localhost:8000/html/nicolas"
    curl -X GET "http://localhost:8000/text"
    curl -X PUT "http://localhost:8000/get-or-create-task/error"
    curl -X PUT "http://localhost:8000/get-or-create-task/foo2"
    curl -X PUT "http://localhost:8000/get-or-create-task/foo"
    curl -X GET "http://localhost:8000/json_any"
    curl -X GET "http://localhost:8000/json_model"
    curl -X GET "http://localhost:8000/download"
    curl -o downloaded_file.txt http://localhost:8000/download
    curl -o downloaded_file_2.txt http://localhost:8000/download_2


Basic bearer token implementation
*********************************

.. code-block:: python

    from fastapi import FastAPI, Depends, status
    from starlette.requests import Request
    from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
    from fastapi.exceptions import HTTPException
    from typing import Optional


    app = FastAPI()


    class BearerTokenChecker(HTTPBearer):
        def __init__(self, token: str):
            super().__init__()
            self.token = token

        async def __call__(self, request: Request) -> Optional[HTTPAuthorizationCredentials]:
            auth = await super().__call__(request)
            if auth is None:
                return None
            if auth.credentials != self.token:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail="Invalid authentication credentials",
                )
            return auth


    security = BearerTokenChecker(token="super_secret_token_value")


    @app.get('/secret', dependencies=[Depends(security)])
    def get_secret():
        return "I love you"


You can test the route using the :code:`curl` command below:

.. code-block:: bash

    export API_AUTH_TOKEN=super_secret_token_value
    curl -X GET -H "Authorization: Bearer $API_AUTH_TOKEN" "http://localhost:8000/secret"


Socket.IO
#########


For the test below, I choose the most popular Socket.IO module for flask: :code:`fastapi-socketio` (Github: https://github.com/pyropy/fastapi-socketio)
You can install it using the command: :code:`pip install fastapi-socketio`

.. code-block:: python

    from fastapi import FastAPI
    from fastapi_socketio import SocketManager

    app = FastAPI()
    socket_manager = SocketManager(app=app, mount_location="/")


    @app.sio.on('Custom Event')
    async def test_message(sid, message):
        print(f"Message received: {message}")
        # By default, message are sent to the same client from which it received the message
        await app.sio.emit('Response FastAPI SocketIO', "Et voila ! Happy ?")
        # In 'emit' you can send a message to another client using argument 'to=other_client_sid'

    @app.sio.on('my broadcast event')
    async def test_message_broadcast(sid, message):
        # With broadcast=True, the message is sent to all connected clients
        await app.sio.emit('Response FastAPI SocketIO', {'data': message['data']}, broadcast=True)

    @app.sio.on('connect')
    async def handle_join(sid, *args, **kwargs):
        await app.sio.emit('Response FastAPI SocketIO', {'data': 'Connected'})
        print(f'Client with sid "{sid}" connected')

    @app.sio.on('disconnect')
    async def handle_leave(sid, *args, **kwargs):
        print(f'Client with sid "{sid}" disconnected')


    if __name__ == '__main__':
        import uvicorn

        uvicorn.run("simple_api:app", host='0.0.0.0', port=8000, reload=True)


You can test the routes using the :code:`python` code below:

.. code-block:: python

    import socketio


    sio = socketio.Client()

    @sio.event
    def connect():
        print('Connected to server')
        sio.emit('Custom Event', {'data': 'Hello from client'})

    @sio.event
    def disconnect():
        print('Disconnected from server')

    @sio.on('Response Flask SocketIO')
    def on_message(message):
        print(f"Received response: {message}")

    sio.connect('http://localhost:5000')
    sio.wait()


Rest API with OAuth & MFA
#########################

In this project, I provide an example of a REST FastAPI app that implements commonly used features following best practices.

Rest Flask Template: [WORK IN PROGRESS]


------------------------------------------------------------

**Sources**:

- FastAPI documentation: https://fastapi.tiangolo.com
- fastapi-socketio documentation: https://github.com/pyropy/fastapi-socketio