Flask Integration

FlexiQ(app) extension with config-driven Queue setup and CLI commands.

flexiq provides a first-class Flask extension that configures a Queue from your app config and registers CLI commands.

Installation

bash

pip install flexiq[flask]

Basic setup

python

from flask import Flask
from flexiq.contrib.flask import FlexiQ

app = Flask(__name__)
app.config["FLEXIQ_DB_PATH"] = "myapp.db"

flexiq = FlexiQ(app)

@flexiq.queue.task()
def send_email(to: str, subject: str):
    ...

Factory pattern

python

from flexiq.contrib.flask import FlexiQ

flexiq = FlexiQ()

def create_app():
    app = Flask(__name__)
    app.config["FLEXIQ_DB_PATH"] = "myapp.db"
    flexiq.init_app(app)
    return app

Configuration

All configuration is read from app.config:

Config Key	Default	Description
`FLEXIQ_DB_PATH`	`.flexiq/flexiq.db`	SQLite database path
`FLEXIQ_BACKEND`	`"sqlite"`	Storage backend: `"sqlite"` or `"postgres"`
`FLEXIQ_DB_URL`	`None`	PostgreSQL connection URL (when backend is `"postgres"`)
`FLEXIQ_WORKERS`	`0` (auto)	Number of worker threads
`FLEXIQ_SCHEMA`	`"flexiq"`	PostgreSQL schema name
`FLEXIQ_DEFAULT_RETRY`	`3`	Default retry count for tasks
`FLEXIQ_DEFAULT_TIMEOUT`	`300`	Default timeout in seconds
`FLEXIQ_DEFAULT_PRIORITY`	`0`	Default task priority
`FLEXIQ_RESULT_TTL`	`None`	Auto-purge completed jobs after N seconds
`FLEXIQ_DRAIN_TIMEOUT`	`30`	Seconds to wait for running tasks on shutdown

Using FLEXIQ_BACKEND = "postgres" requires the flexiq[postgres] extra (pip install flexiq[postgres]) — see the PostgreSQL guide.

Extension options

The FlexiQ constructor accepts a cli_group parameter to rename the CLI command group:

python

# Commands will be under `flask tasks worker`, `flask tasks info`, etc.
flexiq = FlexiQ(app, cli_group="tasks")

CLI commands

The extension registers commands under the flask flexiq group (configurable via cli_group):

`flask flexiq worker`

Start a flexiq worker:

bash

flask flexiq worker
flask flexiq worker --queues default,emails

`flask flexiq info`

Show queue statistics. Supports --format table (default) and --format json:

bash

flask flexiq info
flask flexiq info --format json

code

flexiq queue statistics
------------------------------
  pending      12
  running      3
  completed    450
  failed       2
  dead         1
  cancelled    0
------------------------------
  total        468

Accessing the queue

python

# Via the extension instance
flexiq.queue.stats()

# Via app extensions
app.extensions["flexiq"].queue.stats()

Full example

A complete Flask application with task definitions and routes:

python

from flask import Flask, jsonify, request
from flexiq.contrib.flask import FlexiQ

app = Flask(__name__)
app.config["FLEXIQ_DB_PATH"] = "myapp.db"
app.config["FLEXIQ_DEFAULT_RETRY"] = 3
app.config["FLEXIQ_RESULT_TTL"] = 86400  # 24h auto-cleanup

flexiq = FlexiQ(app)

@flexiq.queue.task()
def send_welcome_email(user_id: int):
    """Send a welcome email to a new user."""
    user = get_user(user_id)
    send_email(user.email, "Welcome!", "Thanks for signing up.")

@flexiq.queue.task(rate_limit="10/m")
def generate_report(report_type: str, params: dict):
    """Generate a report (rate-limited to 10/minute)."""
    return create_report(report_type, params)

@app.route("/api/users", methods=["POST"])
def create_user():
    user = create_user_in_db(request.json)
    send_welcome_email.delay(user.id)
    return jsonify({"id": user.id}), 201

@app.route("/api/reports", methods=["POST"])
def request_report():
    job = generate_report.delay(
        request.json["type"],
        request.json.get("params", {}),
    )
    return jsonify({"job_id": job.id}), 202

@app.route("/api/reports/<job_id>")
def report_status(job_id: str):
    job = flexiq.queue.get_job(job_id)
    if job is None:
        return jsonify({"error": "Not found"}), 404
    return jsonify({"status": job.status, "progress": job.progress})

@app.route("/api/queue/stats")
def queue_stats():
    return jsonify(flexiq.queue.stats())

Run the app and worker:

bash

# Terminal 1: Flask app
flask run

# Terminal 2: Worker
flask flexiq worker