Store and monitor site traffic.
pip install flask-trafficfrom flask import Flask
from flask_traffic import Traffic
from flask_traffic.stores import JSONStore
traffic = Traffic()
def create_app():
app = Flask(__name__)
json_store = JSONStore()
traffic.init_app(app, stores=json_store)
@app.route('/')
def index():
return 'Hello, World!'
@app.route('/traffic')
def traffic():
return json_store.read()
return appinstance/traffic.json
...
{
"request_date": "2024-12-03T20:10:34.932025",
"request_method": "GET",
"request_path": "/",
"request_remote_address": "127.0.0.1",
"request_referrer": null,
"request_user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
"request_browser": null,
"request_platform": null,
"response_time": 1,
"response_size": 13,
"response_status_code": 200,
"response_exception": null,
"response_mimetype": "text/html"
}
...
from flask_traffic import LogPolicy
The log policy is used to tell Flask-Traffic what data to store after a request is made in whatever store, or stores you have configured.
A new instance of LogPolicy will have all the log attributes set to True by
default.
You can use the methods set_from_true or set_from_false to select which attributes
to store.
set_from_true will allow you to disable certain attributes from being stored.
set_from_false will allow you to enable certain attributes to be stored.
If a store is created without a log policy passed in, one is created with all log
attributes set to True.
only_on_exception, and skip_on_exception are set to False.
on_endpoints, skip_endpoints, on_status_codes, and skip_status_codes are
used to scope or skip logging based on the endpoint or status code. These are disabled
by default.
Here's an example of the LogPolicy class only storing the date and request path:
from flask_traffic.stores import JSONStore
from flask_traffic import LogPolicy
log_policy = LogPolicy().set_from_false(
request_date=True,
request_path=True
)
json_store = JSONStore(log_policy=log_policy)Results in:
...
{
"request_date": "2024-12-03T20:33:43.051597",
"request_path": "/"
}
...
Here's an example of the LogPolicy class storing everything except the response size:
from flask_traffic.stores import JSONStore
from flask_traffic import LogPolicy
log_policy = LogPolicy().set_from_true(
response_size=False
)
json_store = JSONStore(log_policy=log_policy)This store saves traffic data in a JSON file. The file is created in the
instance folder of the Flask app by default.
This store saves traffic data in a CSV file. The file is created in the
instance folder of the Flask app by default.
This store saves traffic data in a SQL type database. It defaults to using SQLite
which is created in the instance folder of the Flask app by default.
You can specify a database URL, or pass in an already created SQLAlchemy engine.
This store is used if you want to store traffic data in a SQL type database.
This is an ORM version of the SQLStore. It is designed to integrate with an existing
SQLAlchemy ORM environment like Flask-SQLAlchemy.
This mixin is used to set the correct table columns for the SQLORMStore.
Example:
from flask_traffic.stores import SQLORMModelMixin
from app import db
class Traffic(db.Model, SQLORMModelMixin):
passEach store has a read method that will return the data in the store as a
list of dictionaries.
Here's an example of reading the data from a CSVStore:
@app.route("/read-csv")
def read_csv():
return csv_store.read()This will return the data as JSON.
You can also override the read method to change the default,
or add more methods of course.
here's an example of overriding the read method in a SQLStore to only return data
where the response status code is 200:
class MyStore(SQLStore):
def read(self):
with self.database_engine.connect() as connection:
results = connection.execute(
self.database_log_table.select().order_by(
self.database_log_table.c.traffic_id.desc()
).where(
self.database_log_table.c.response_status_code == 200
)
)
return [row._asdict() for row in results.fetchall()]This example will store traffic data in a SQL database using Flask-SQLAlchemy and store any traffic that causes exceptions in a JSON file.
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_traffic import Traffic, LogPolicy
from flask_traffic.stores import JSONStore, SQLORMStore, SQLORMModelMixin
db = SQLAlchemy()
traffic = Traffic()
class Cars(db.Model):
car_id = db.Column(db.Integer, primary_key=True)
make = db.Column(db.String(80), unique=True, nullable=False)
model = db.Column(db.String(80), unique=True, nullable=False)
class Traffic(db.Model, SQLORMModelMixin):
pass
def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///instance/db.sqlite'
db.init_app(app)
# init traffic after db.init_app to find the db session
traffic.init_app(
app, stores=[
JSONStore(
log_policy=LogPolicy(only_on_exception=True)
),
SQLORMStore(
Traffic,
log_policy=LogPolicy(skip_on_exception=True)
)
])
@app.route('/')
def index():
return 'Hello, World!'
return appThis example will store traffic data in a CSV file and only store the IP address
from flask import Flask
from flask_traffic import Traffic, LogPolicy
from flask_traffic.stores import CSVStore
traffic = Traffic()
def create_app():
app = Flask(__name__)
traffic.init_app(
app,
stores=CSVStore(
log_policy=LogPolicy().set_from_false(request_remote_address=True)
)
)
@app.route('/')
def index():
return 'Hello, World!'
return app