Script en flask para registrar cada request (para pruebas)

Hoy me aventuré por un camino inusual. Mi día comenzó con el objetivo de guardar mis tweets marcados y mis likes en una base de datos. Para lograrlo, necesitaba acceder al API de Twitter usando tweepy. Sin embargo, me encontré con un obstáculo persistente: Twitter requiere una URL de autorización para probar su API, lo cual complicaba mis intentos de configurar ngrok con un pequeño servidor Python debido a problemas de CSRF.

Después de un tiempo de frustración, tuve una idea: "Tengo mi propio dominio, ¿por qué no añadir un subdominio y utilizarlo como URL de autorización?"

Requerimientos

Disponer de un dominio que pueda recibir peticiones
Mostrar todos los requests y sus parámetros en una URL específica (/dashboard)
Mantener la información persistente utilizando SQLite

Script

Con la ayuda de Skynet, obtuve esto:

::python
import sqlite3
from flask import Flask, request, jsonify, render_template_string

app = Flask(__name__)
DATABASE = 'requests_log.db'
MAX_REQUESTS = 1000
DELETE_BATCH_SIZE = 500  # Number of records to delete when exceeding MAX_REQUESTS


def init_db():
    """Initialize the SQLite database with requests table."""
    with sqlite3.connect(DATABASE) as conn:
        cursor = conn.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS requests (
                            id INTEGER PRIMARY KEY AUTOINCREMENT,
                            method TEXT,
                            path TEXT,
                            params TEXT,
                            headers TEXT,
                            body TEXT
                        )''')
        conn.commit()


def log_request(method, path, params, headers, body):
    """Log request details into SQLite database, excluding '/dashboard'."""
    if path != 'dashboard':  # Exclude logging for '/dashboard' endpoint
        with sqlite3.connect(DATABASE) as conn:
            cursor = conn.cursor()
            cursor.execute('INSERT INTO requests (method, path, params, headers, body) VALUES (?, ?, ?, ?, ?)',
                           (method, path, params, headers, body))
            conn.commit()

            # Check the number of requests and delete oldest if exceeds MAX_REQUESTS
            cursor.execute('SELECT COUNT(*) FROM requests')
            count = cursor.fetchone()[0]
            if count > MAX_REQUESTS:
                cursor.execute('DELETE FROM requests WHERE id IN (SELECT id FROM requests ORDER BY id ASC LIMIT ?)',
                               (DELETE_BATCH_SIZE,))
                conn.commit()


# Initialize the database when the app starts
init_db()


@app.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE'])
@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
def catch_all(path):
    """Route to catch all requests and log them."""
    method = request.method
    params = request.args.to_dict()
    headers = dict(request.headers)
    body = request.data.decode('utf-8')

    log_request(method, path, str(params), str(headers), body)

    return 'Request logged', 200


@app.route('/dashboard', methods=['GET'])
def dashboard():
    """Route to display request logs in a dashboard."""
    with sqlite3.connect(DATABASE) as conn:
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM requests')
        rows = cursor.fetchall()
        requests_list = [
            {'id': row[0], 'method': row[1], 'path': row[2], 'params': row[3], 'headers': row[4], 'body': row[5]} for
            row in rows]

    # HTML template to render the dashboard view
    table_template = """
    <!doctype html>
    <html lang="en">
      <head>
        <meta charset="utf-8">
        <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
        <title>Request Log Dashboard</title>
        <style>
            table {
                width: 100%;
                border-collapse: collapse;
            }
            table, th, td {
                border: 1px solid black;
            }
            th, td {
                padding: 8px;
                text-align: left;
            }
            th {
                background-color: #f2f2f2;
            }
        </style>
      </head>
      <body>
        <h1>Request Log Dashboard</h1>
        <table>
          <tr>
            <th>ID</th>
            <th>Method</th>
            <th>Path</th>
            <th>Params</th>
            <th>Headers</th>
            <th>Body</th>
          </tr>
          {% for request in requests %}
          <tr>
            <td>{{ request.id }}</td>
            <td>{{ request.method }}</td>
            <td>{{ request.path }}</td>
            <td>{{ request.params }}</td>
            <td>{{ request.headers }}</td>
            <td>{{ request.body }}</td>
          </tr>
          {% endfor %}
        </table>
      </body>
    </html>
    """
    return render_template_string(table_template, requests=requests_list)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

En principio la lógica es muy simple:

Todo request (que nos sea el /dashboard) lo registra en una base de datos sqlite. Teniendo en cuenta que si se excede cierta cantidad de 1000 requests estoy eliminando los 500 más antiguos.
Al consultar el dashboard, me lista los requests con un formato html mínimamente cómodo.

Claro, para correrlo en un servidor tuve que utilizar supervisord, uwsgi y nginx, aunque quizá se pudo realizar con menos.

Mi archivo uwsgi.ini terminó así:

[uwsgi]
webapps = /home/alexander/webapps
app = %(webapps)/params.alexanderae.com
venv = /home/alexander/.pyenv/versions/flask

chdir = %(app)
module = server:app
virtualenv = %(venv)

uwsgi-socket = %(app)/uwsgi.sock
master = true
processes = 2
threads = 2
thread-stacksize = 128
reload-on-rss = 30
harakiri = 10
vacuum = true

logto = /tmp/uwsgi_params.log

Y mi script de supervisor:

[program:server_app]
command=/home/alexander/.pyenv/versions/flask/bin/uwsgi /home/alexander/webapps/params.alexanderae.com/uwsgi.ini
autostart=true
autorestart=true
stderr_logfile=/var/log/flask_app.err.log
stdout_logfile=/var/log/flask_app.out.log
user=alexander

Y con ello parece que pasaré al paso 2, el cual es intentar conectarme a twitter, esperemos y funcione, ya lo iré contando (o tal vez no).

Script en flask para registrar cada request (para pruebas)

Requerimientos

Script

Comentarios !