Flask

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

Using python venv:

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:

python3 -m flask --app simple_api run
# OR
flask --app simple_api run

You can also use docker to run the APP:

  1. Create the 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"]

  2. Save the code to test to the file simple_api.py

  3. Run the Dockerfile:

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

Simple API

Receiving data from the client

from flask import Flask, request


app = Flask(__name__)

# Query parameters
@app.route('/query', methods=['GET'])
def query_paramters():
    key_value = request.args.get('key', '')
    print(f"key: {key_value}")
    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=['POST'])
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')
    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 curl commands below:

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 POST "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

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


app = Flask(__name__)

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

@app.route('/tuple')
def tuple_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 curl commands below:

curl -X GET "http://localhost:5000/html_txt/nicolas"
curl -X GET "http://localhost:5000/tuple"
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

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 curl command below:

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

WebSockets

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

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(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)

flask-socketio also has the concept of 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 python code below:

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: