FAQ

Frequently asked questions about flexiq.

Can I use flexiq with Django?

Yes. Create a Queue instance in one of your Django apps and import it where needed:

python

# myproject/tasks.py
from flexiq import Queue

queue = Queue(db_path="flexiq.db")

@queue.task()
def send_welcome_email(user_id: int):
    from myapp.models import User
    user = User.objects.get(id=user_id)
    user.email_user("Welcome!", "Thanks for signing up.")

Import tasks lazily inside the function body to avoid Django app registry issues. Start the worker separately:

bash

DJANGO_SETTINGS_MODULE=myproject.settings flexiq worker --app myproject.tasks:queue

Can I use flexiq with Flask?

Yes. Same pattern — define a queue, decorate tasks, run the worker:

python

# tasks.py
from flexiq import Queue

queue = Queue(db_path="flexiq.db")

@queue.task()
def generate_report(report_id: int):
    from myapp import create_app
    app = create_app()
    with app.app_context():
        ...

Can I use flexiq with Express, Fastify, or NestJS?

Yes — all three. Define the queue and tasks once, then mount the REST API and dashboard with the matching contrib helper. The worker runs the same way regardless of framework; each helper's peer (express / fastify / @nestjs/common) is installed separately.

Express — @byteveda/flexiq/contrib/express:

import express from "express";
import { flexiqRouter, flexiqDashboard } from "@byteveda/flexiq/contrib/express";

const app = express();
app.use("/tasks", flexiqRouter(queue));     // REST API
app.use("/admin", flexiqDashboard(queue));  // dashboard SPA
app.listen(3000);

Fastify — @byteveda/flexiq/contrib/fastify:

import Fastify from "fastify";
import { flexiqFastify, flexiqDashboardPlugin } from "@byteveda/flexiq/contrib/fastify";

const app = Fastify();
await app.register(flexiqFastify, { queue, prefix: "/tasks" });
await app.register(flexiqDashboardPlugin, { queue, prefix: "/admin" });
await app.listen({ port: 3000 });

NestJS — @byteveda/flexiq/contrib/nest:

import { Module } from "@nestjs/common";
import { FlexiQModule } from "@byteveda/flexiq/contrib/nest";

@Module({ imports: [FlexiQModule.forRoot(queue)] })
export class AppModule {}

Inject FlexiQService into your providers to enqueue and read results.

Can I use flexiq with Spring Boot?

Yes, via the org.byteveda:flexiq-spring starter (Spring Boot 3). Add it and the auto-configuration builds a FlexiQ bean from your flexiq.* properties — inject it anywhere:

java

@Service
public class SignupService {
    private final FlexiQ flexiq;
    public SignupService(FlexiQ flexiq) { this.flexiq = flexiq; }

    public void register(User user) {
        flexiq.enqueue("send_welcome_email", user.email());
    }
}

yaml

flexiq:
  url: postgres://localhost/flexiq
  pool-size: 8
  namespace: my-app

The worker is not auto-started — build one from the injected bean in an ApplicationRunner, or set flexiq.dashboard.enabled=true to auto-start the dashboard. Non-Spring apps just build a FlexiQ and Worker programmatically.

Yes, with caveats. SQLite in WAL mode allows concurrent readers and one writer at a time. flexiq sets busy_timeout=5000ms to handle contention.

However, flexiq is designed as a single-process task queue. Multiple worker processes against one database works but will see diminishing returns due to write lock contention. For most workloads, one worker process with multiple threads is sufficient.

What happens if my worker crashes mid-task?

The job stays in running status in SQLite. On the next worker start, the stale job reaper detects jobs that have been running longer than their timeout and marks them as failed (triggering retries or DLQ).

If no timeout is set, stale jobs remain in running status indefinitely. Always set a timeout on your tasks.

python

@queue.task(timeout=300)  # 5 minute timeout
def process(data):
    ...

queue.task("process", process, { timeoutMs: 300_000 }); // 5 minute timeout

java

Task<Data> process = Task.of("process", Data.class)
        .timeout(Duration.ofMinutes(5));

How big can the SQLite database get?

SQLite can handle databases up to 281 TB (theoretical limit). In practice, flexiq databases stay small if you purge finished jobs. Without cleanup, expect ~1 KB per job — a million completed jobs ≈ 1 GB.

Set result_ttl to auto-purge old jobs:

python

queue = Queue(db_path="myapp.db", result_ttl=86400)  # Purge after 24h

There is no result-TTL knob; call purgeCompleted on a schedule (the argument is milliseconds):

setInterval(() => queue.purgeCompleted(86_400_000), 3_600_000); // keep 24h

There is no result-TTL knob; call purgeCompleted on a schedule (the argument is milliseconds):

java

flexiq.purgeCompleted(Duration.ofHours(24).toMillis()); // keep 24h

Can I use a remote or networked SQLite?

No. SQLite requires local filesystem access for file locking. Network filesystems (NFS, SMB, CIFS) do not reliably support the locking primitives SQLite depends on. Always place the database on local storage.

When should I use Postgres instead of SQLite?

Use the Postgres backend when you need:

Multi-machine workers — run workers on separate servers sharing the same queue
Higher write throughput — Postgres handles concurrent writers better than SQLite
Existing Postgres infrastructure — leverage your existing database and backups

For single-machine workloads, SQLite is simpler and requires zero setup.

Install the extra, then point the queue at Postgres:

python

# pip install flexiq[postgres]
queue = Queue(db_url="postgres://localhost/flexiq")

Postgres is already compiled into the prebuilt native binary — no extra install, just switch the backend:

new Queue({ backend: "postgres", dsn: process.env.PG_URL, schema: "flexiq" });

Postgres ships inside the bundled native library — no extra dependency, just switch the backend:

java

FlexiQ.builder().postgres(System.getenv("PG_URL")).open();

See the Postgres backend guide.

Is flexiq production-ready?

flexiq is suitable for production workloads — background job processing, periodic tasks, data pipelines, and similar use cases.

For single-machine deployments, use the default SQLite backend. For multi-server setups, use the Postgres backend.

What observability options does flexiq support?

flexiq offers three observability integrations, each implemented as a TaskMiddleware and combinable:

Integration	Best for	Install
OpenTelemetry	Distributed tracing, correlating tasks with HTTP requests	`pip install flexiq[otel]`
Prometheus	Metrics dashboards, alerting on queue depth/error rates	`pip install flexiq[prometheus]`
Sentry	Error tracking with rich context and breadcrumbs	`pip install flexiq[sentry]`

flexiq ships three observability middlewares under @byteveda/flexiq/contrib/*. Install the peer you use, then register with queue.use(...):

Integration	Best for	Peer
OpenTelemetry	Distributed tracing, correlating tasks with HTTP requests	`@opentelemetry/api`
Prometheus	Metrics dashboards, alerting on queue depth/error rates	`prom-client`
Sentry	Error tracking with rich context and breadcrumbs	`@sentry/node`

import { otelMiddleware } from "@byteveda/flexiq/contrib/otel";
queue.use(otelMiddleware());

flexiq ships two contrib middlewares, registered with flexiq.use(...). Their third-party dependency is compileOnly — add the runtime dep yourself:

Integration	Best for	Runtime dep
Micrometer (`FlexiQObservation`)	One instrumentation → both a metrics timer and a trace span	`io.micrometer:micrometer-observation`
Sentry (`SentryMiddleware`)	Error tracking on failed attempts and dead-letters	`io.sentry:sentry`

java

flexiq.use(new FlexiQObservation(registry));

OpenTelemetry and Prometheus are reached downstream of Micrometer — OTel as a micrometer-tracing backend, Prometheus as the dashboard's /metrics endpoint or a Micrometer MeterRegistry. There is no dedicated OTel/Prometheus class.

How does flexiq compare to running Celery with SQLite?

Celery can use SQLite as a result backend, but still requires a broker (Redis or RabbitMQ). flexiq replaces both broker and backend with a single SQLite database. Additionally:

flexiq's scheduler runs in Rust (faster polling, lower overhead)
Worker threads are OS threads managed by Rust, not Python processesseparate Node processesseparate JVM processes
No external dependencies beyond cloudpicklethe native addonthe bundled native library

Can I use async tasks?

Yes. Define the task function with async defasynca handler that runs on the worker's executor and the worker dispatches it natively — no asyncio.run() wrapping, no thread-pool bridgingeach runs on your Node event loop, no thread per jobeach job runs as its own executor task, sized by concurrency(n) or autoscale:

python

@queue.task()
async def fetch_urls(urls: list[str]) -> list[str]:
    import httpx
    async with httpx.AsyncClient() as client:
        return [r.text for r in await asyncio.gather(
            *[client.get(url) for url in urls]
        )]

Enqueue and await results from async application code:

python

job = fetch_urls.delay(urls)
result = await job.aresult(timeout=30)
stats = await queue.astats()

queue.task("fetchUrls", async (urls: string[]) =>
  Promise.all(urls.map((u) => fetch(u).then((r) => r.text()))),
);

Enqueue and await results from your application code:

const id = queue.enqueue("fetchUrls", [urls]);
const result = await queue.result(id);
const stats = queue.stats();

java

Task<List<String>> fetchUrls = Task.of("fetchUrls", new TypeReference<>() {});
List<String> urls = List.of("https://example.com/a", "https://example.com/b");

try (Worker worker = queue.worker()
        .handle(fetchUrls, batch -> batch.stream().map(App::fetch).toList())
        .concurrency(8)
        .start()) {
    String id = queue.enqueue(fetchUrls, urls);
    queue.awaitJob(id, Duration.ofSeconds(30));
    // getResult takes a Class token, so read the JSON array as String[].
    String[] result = queue.getResult(id, String[].class).orElseThrow();
}

Sync and async tasks can coexist in the same queue. The worker automatically routes each job to the correct pool based on the task type. See the Async Tasks guide for details including concurrency tuning and

current_jobcurrentJob()middleware-provided job context

context in async tasks.

What serialization format does flexiq use?

By default, SmartSerializerJsonSerializerJsonSerializer — which uses MessagePack with a cloudpickle fallback, so it handles most Python objects including lambdas and closures; switch to JsonSerializer for simpler, cross-language payloadshuman-readable JSON bytes; switch to MsgpackSerializer for compact, binary payloadshuman-readable JSON bytes via Jackson; switch to MsgpackSerializer for compact, binary payloads — or provide a custom serializer:

python

from flexiq import Queue, JsonSerializer

queue = Queue(serializer=JsonSerializer())

Custom serializers implement the Serializer protocol with dumps(obj) -> bytes and loads(data) -> Any methods.

import { Queue, MsgpackSerializer } from "@byteveda/flexiq";

new Queue({ dbPath: "flexiq.db", serializer: new MsgpackSerializer() });

Custom serializers implement the Serializer interface with serialize(value) -> bytes and deserialize(bytes) -> value methods.

java

FlexiQ queue = FlexiQ.builder()
    .sqlite("flexiq.db")
    .serializer(new MsgpackSerializer())
    .open();

Custom serializers implement the Serializer interface with serialize(value) and deserialize(bytes, type) methods.

Regardless of serializer, avoid passing unserializable objects like open file handles, database connections, or thread locks.

Can I run the dashboard and worker in the same process?

They're designed to run as separate processes sharing the same database.

bash

# Terminal 1
flexiq worker --app myapp:queue

# Terminal 2
flexiq dashboard --app myapp:queue

For embedding in a FastAPI app, use FlexiQRouter instead — it provides the same stats and job management as REST endpoints.

bash

# Terminal 1 — run a worker over a module that exports a Queue
flexiq run ./app.js --queues default

# Terminal 2 — serve the dashboard
flexiq --db flexiq.db dashboard --port 8787

To embed the dashboard in an existing HTTP server, call serveDashboard(queue, { port }).

Workers are started programmatically (there is no flexiq worker command); the dashboard has both a programmatic API and a CLI command:

java

// In your app — start a worker
Worker worker = flexiq.worker().handle(myTask, ...).concurrency(4).start();

// Anywhere — start the dashboard
flexiq.dashboard(8081); // or DashboardServer.start(flexiq, 8081)

How do I reset / clear all jobs?

python

# Purge all completed jobs
queue.purge_completed(older_than=0)

# Purge all dead letters
queue.purge_dead(older_than=0)

// Purge all completed jobs and dead letters (argument is milliseconds; 0 = all)
await queue.purgeCompleted(0);
await queue.purgeDead(0);

java

// Purge all completed jobs and dead letters (argument is milliseconds; 0 = all)
flexiq.purgeCompleted(0L);
flexiq.purgeDead(0L);

Or delete the database file and restart:

bash

rm myapp.db myapp.db-wal myapp.db-shm