Flask
=====


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

Using python venv:

    .. code-block:: bash

        python3 -m venv venv-flask
        source venv-flask/bin/activate
        pip install flask flask-socketio websocket-client requests


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

    .. code-block:: bash

        python3 -m flask --app simple_api run
        # OR
        flask --app simple_api run
    
    .. note::
        The default port used is :code:`5000`, you can change it using the argument :code:`--port` (for example: :code:`flask --app simple_api run --port 8000`)

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 flask flask-socketio websocket-client requests
            CMD ["python3", "-m", "flask", "--app", "simple_api", "run"]

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

    #. Run the :code:`Dockerfile`:

        .. code-block:: bash

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


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

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

.. code-block:: python

    from flask import Flask, request


    app = Flask(__name__)

    # Query parameters
    @app.route('/query', methods=['GET'])
    def query_paramters():
        p1 = request.args['p1']
        p2 = request.args.get('p2', '')
        print(f"p1: {p1}")
        print(f"p2 (optional): {p2}")
        return ""

    # Path parameters
    @app.route('/path/<username>/<int:post_id>', methods=['POST'])
    def path_paramters(username, post_id):
        print(f'username: {username}, post_id: {post_id}')
        return ""

    # Request body parameters
    @app.route('/body_raw', methods=['POST'])
    def body_raw_parameters():
        raw_data = request.get_data()
        print(f'raw_data: {raw_data}')
        return ""

    @app.route('/body_json', methods=['POST'])
    def body_json_parameters():
        json_data = request.get_json()
        print(json_data)
        return ""

    # HTTP headers parameters
    # List of headers: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers
    @app.route('/header', methods=['GET'])
    def header_parameters():
        user_agent = request.headers.get('User-Agent')
        print(user_agent)
        return ""

    # Cookies parameters
    @app.route('/cookies')
    def cookies_parameters():
        username = request.cookies.get('username')
        print(username)
        return ""

    # Form Data (URL-encoded or Multipart)
    @app.route('/form', methods=['POST'])
    def form_parameters():
        # Works with content-type 'application/x-www-form-urlencoded' OR 'multipart/form-data'
        name = request.form.get('name')
        email = request.form.get('email')
        print(f"Content-Type: {content_type}")
        if 'multipart/form-data' in request.content_type:
            f = request.files.get('file_in')
            print(f"Content of 'file_in': {f.read()}")
            f.seek(0)
            f.save('uploaded_file.txt')
        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:5000/query?key=value123"
    curl -X POST "http://localhost:5000/path/nicoh/101"
    curl -X POST "http://localhost:5000/body_raw" -d "This is raw data"
    curl -X POST "http://localhost:5000/body_json" -H "Content-Type: application/json" -d '{"name": "Hania <3", "age": 30}'
    curl -X GET "http://localhost:5000/header" -H "User-Agent: CustomUserAgent/1.0"
    curl -X GET "http://localhost:5000/cookies" --cookie "username=nicoh"
    curl -X POST "http://localhost:5000/form" -d "name=Nicolas&email=nicolas@example.com"
    echo "Coucou" > file_test.txt
    curl -X POST "http://localhost:5000/form" -F "name=Nicolas" -F "email=nicolas@example.com" -F "file_in=@file_test.txt"


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

.. code-block:: python

    from flask import Flask, jsonify, send_file
    from markupsafe import escape
    import io


    app = Flask(__name__)

    # Text / HTML response
    @app.route('/html/<name>')
    def html_response(name):
        # escape: To protect from injection attacks
        return f"<h1>Hello {escape(name)}</h1>"

    @app.route('/text')
    def text_response():
        # (Content, Status Code, Headers)
        return "This is the content", 200, {'Content-Type': 'text/plain'}

    # json response
    @app.route('/json')
    def json_response():
        data = {"message": "Hello, World!", "status": "success"}
        return jsonify(data)
        # Equivalent to:
        # return Response(
        #     response=json.dumps(data),
        #     status=200,
        #     mimetype='application/json'
        # )

    # File
    @app.route('/download')
    def download_file():
        file_stream = io.BytesIO()
        file_stream.write(b'This is some file data.')
        file_stream.seek(0)
        return send_file(file_stream, as_attachment=True, download_name='output.txt')


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

.. code-block:: bash

    curl -X GET "http://localhost:5000/html/nicolas"
    curl -X GET "http://localhost:5000/text"
    curl -X GET "http://localhost:5000/json"
    curl -X GET "http://localhost:5000/download"
    curl -o downloaded_file.txt http://localhost:5000/download


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

.. code-block:: python

    from functools import wraps
    from flask import Flask, request, jsonify


    def check_authorization_token(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            auth_header = request.headers.get('Authorization')
            if not auth_header:
                return jsonify({"message": "Authorization header missing"}), 401
            if not auth_header.startswith("Bearer "):
                return jsonify({"message": "Invalid authorization type, expected Bearer"}), 400
            token = auth_header.split(" ")[1]
            if token != "super_secret_token_value":
                return jsonify({"message": "Invalid token"}), 401
            return func(*args, **kwargs)
        return wrapper


    app = Flask(__name__)


    @app.route('/secret', methods=['GET'])
    @check_authorization_token
    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:5000/secret"


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

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


.. code-block:: python

    from flask import Flask, request
    from flask_socketio import SocketIO, emit


    app = Flask(__name__)
    socketio = SocketIO(app)

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

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

    @socketio.on('connect')
    def handle_connect():
        print(f'Client with sid "{request.sid}" connected')
        emit('Response Flask SocketIO', {'data': 'Connected'})

    @socketio.on('disconnect')
    def test_disconnect():
        print(f'Client with sid "{request.sid}" disconnected')

    if __name__ == '__main__':
        socketio.run(app)


:code:`flask-socketio` also has the concept of :code:`room` , you can find more information in the official documentation: https://flask-socketio.readthedocs.io/en/latest/getting_started.html#rooms


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 Flask app that implements commonly used features following best practices.


Rest Flask Template: https://gitlab.com/ndejax/rest-flask-template


An overview of the project environment:

- Database: Postgresql with SqlAlchemy
- WSGI Server: gunicorn
- Access/Refresh token: flask_jwt_extended
- Two Factor Authentication: pyotp
- Background tasks with Celery (+ RabbitMQ broker)
- Ready-to-use docker compose file
- Dependency managment/packaging: pyproject.toml comptible with PDM & PIP


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

**Sources**:

- https://flask.palletsprojects.com/en
- https://blog.miguelgrinberg.com/post/easy-websockets-with-flask-and-gevent