Skip to content

Instantly share code, notes, and snippets.

@alexander-ae
Last active June 23, 2024 02:08
Show Gist options
  • Save alexander-ae/2b363883a5ec4fc85b385b220697bbec to your computer and use it in GitHub Desktop.
Save alexander-ae/2b363883a5ec4fc85b385b220697bbec to your computer and use it in GitHub Desktop.
Flask script to log every request
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment