craft-easy-jobs CLI (standalone)¶

craft-easy-jobs is a lightweight standalone package that provides a CLI and runner for Craft Easy-style jobs. Unlike the API job framework, it has no database, no persistent schedules, and no REST API — it is a click-based executable designed for Cloud Run, GitHub Actions, Kubernetes CronJobs, and ad-hoc manual runs.

pip install craft-easy-jobs

# Optional: install with built-in jobs that talk to craft-easy-api
pip install craft-easy-jobs[full]

The only required dependency is click>=8.0. The [full] extra adds craft-easy-api so the built-in jobs (bi-export, token-cleanup, audit-log-archive) have access to Beanie models.

When to use which¶

Craft Easy has two job systems on purpose — they solve different problems.

Scenario	craft-easy-jobs CLI	API jobs module
Ad-hoc manual runs	Yes	No
Cloud Run / ephemeral containers	Yes	No
Cron scheduling with persistent history	No	Yes
Multi-instance coordination (distributed lock)	No	Yes
Embedded inside a FastAPI service	No	Yes
CI/CD, GitHub Actions, Kubernetes CronJob	Yes	No
Admin UI with schedule management	No	Yes
Real-time progress tracking in a database	No	Yes
Per-tenant or parent-tenant scoping	Yes	Partial (`per_tenant` flag)

Rule of thumb: if something should trigger the job (a cron expression you want the API to respect, an admin clicking "Run now", a chain after a previous job) — use the API module. If the job is run from outside the API (Cloud Scheduler, a cron pod, a human at a terminal) — use the CLI.

The two can coexist. The same job definition can be registered in both systems, and teams often wire fast, interactive jobs through the API module while pushing heavy, isolated jobs (BI exports, large data migrations) to dedicated CLI containers.

Defining jobs¶

# jobs/settlement.py
from craft_easy_jobs import job, JobContext

@job(
    name="settlement",
    description="Monthly revenue settlement",
    timeout_seconds=600,
    retries=2,
    params_schema={
        "period": {"type": "string", "pattern": "^\\d{4}-\\d{2}$"},
    },
)
async def settlement_job(ctx: JobContext):
    period = ctx.params["period"]
    await ctx.log(f"Starting settlement for {period}")
    # ... work ...
    return {"processed": 42, "period": period}

The full decorator reference is in Job Decorators & Context.

Running from the command line¶

After installing, the CLI is available as craft-easy-job:

# Trivial run
craft-easy-job run token-cleanup

# With parameters
craft-easy-job run settlement --params '{"period": "2026-03"}'

# For a specific tenant
craft-easy-job run bi-export \
  --tenant-id tenant_664abc \
  --params '{"target_type": "bigquery", "project_id": "...", "dataset_id": "...", "configs": [...]}'

# Parent-tenant scope (runs for the tenant and all descendants)
craft-easy-job run org-rollup \
  --tenant-id parent_org_123 \
  --scope parent_tenant

# Explicit sub-tenant list
craft-easy-job run rollup \
  --tenant-id parent_123 \
  --scope parent_tenant \
  --sub-tenant-ids child1,child2,child3

# List every registered job and chain
craft-easy-job list

CLI options reference¶

Flag	Purpose
`--params`, `-p`	JSON object of parameters (default: `{}`)
`--tenant-id`, `-t`	Tenant to run the job for
`--scope`, `-s`	`tenant` (default), `parent_tenant`, or `system`
`--sub-tenant-ids`	Comma-separated list of descendant tenants (parent_tenant scope)

craft-easy-job list prints every registered job and every chain, including their timeouts, retries, and scopes — useful for smoke-testing a deployment.

Scopes¶

The standalone CLI has a richer tenant model than the API framework. Every job has a scope that determines how it interacts with the tenant tree:

Scope	Behaviour
`tenant` (default)	Runs for one tenant. `ctx.tenant_id` is set; `ctx.iter_tenant_ids()` returns `[tenant_id]`.
`parent_tenant`	Runs for a parent tenant and all its descendants. `ctx.iter_tenant_ids()` iterates the tenant plus every sub-tenant.
`system`	Runs globally with no tenant context. Used by maintenance jobs like `token-cleanup`.

If craft-easy-api is installed, the CLI auto-resolves sub-tenants by calling craft_easy.core.tenant.get_tenant_tree(). Otherwise, pass them explicitly with --sub-tenant-ids.

@job(name="org-rollup", scope="parent_tenant")
async def rollup(ctx: JobContext):
    for tenant_id in ctx.iter_tenant_ids():
        await rollup_tenant(tenant_id)
    return {"processed": len(ctx.iter_tenant_ids())}

Scope comparison: standalone CLI vs API framework¶

The two systems resolve tenant scope at different layers. This table shows the key differences:

Dimension	API job framework (`craft-easy-api`)	Standalone CLI (`craft-easy-jobs`)
Scope source	Request context — inferred from the authenticated user's session	CLI argument: `--scope tenant\\|parent_tenant\\|system`
Default scope	Sub-tenant from the current request	`tenant` — the value passed to `--tenant-id`
Parent-tenant traversal	Via the `per_tenant` opt-in flag on a resource or job	`--scope parent_tenant` — the job iterates all sub-tenants automatically via `ctx.iter_tenant_ids()`
System-wide scope	Not available — API jobs always run within a tenant boundary	`--scope system` — no tenant filtering applied
Sub-tenant discovery	Automatic — tenant tree is part of the request context	Auto-resolved via `craft_easy.core.tenant.get_tenant_tree()` when API is installed; otherwise supply `--sub-tenant-ids` explicitly
`ctx.tenant_id`	Single tenant ID from the request	Single tenant ID (`tenant` scope) or root parent ID (`parent_tenant` scope)
`ctx.iter_tenant_ids()`	Not available	Returns `[tenant_id]` for `tenant` scope; tenant + all descendants for `parent_tenant` scope

Choose the API framework when a user action triggers the job, you need the admin UI or real-time progress tracking, or the job always operates on exactly one tenant.

Choose the standalone CLI with parent_tenant scope when a Cloud Scheduler trigger should fan out across an entire organisation tree, or a platform operator needs a nightly rollup, BI export, or aggregate report spanning multiple child tenants without looping manually.

Deployment patterns¶

GitHub Actions¶

- name: Nightly BI export
  run: |
    pip install craft-easy-jobs[full]
    craft-easy-job run bi-export \
      --tenant-id ${{ secrets.TENANT_ID }} \
      --params '${{ secrets.BI_EXPORT_PARAMS }}'
  env:
    MONGODB_URI: ${{ secrets.MONGODB_URI }}

Google Cloud Run + Cloud Scheduler¶

FROM python:3.12-slim
RUN pip install craft-easy-jobs[full]
COPY my_jobs.py /app/my_jobs.py
ENV PYTHONPATH=/app
CMD ["craft-easy-job", "run", "month-end-close"]

Point Cloud Scheduler at a Cloud Run job and invoke the image on a cron schedule. The container boots, runs the job, and exits — you pay only for the execution time.

Kubernetes CronJob¶

apiVersion: batch/v1
kind: CronJob
metadata:
  name: token-cleanup
spec:
  schedule: "0 3 * * *"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: cleanup
            image: my-registry/craft-easy-jobs:latest
            command: ["craft-easy-job", "run", "token-cleanup"]
          restartPolicy: OnFailure

Ad-hoc troubleshooting¶

# On a developer machine
source .venv/bin/activate
pip install craft-easy-jobs[full]
export MONGODB_URI="mongodb://localhost:27017"
craft-easy-job run settlement --params '{"period": "2026-03", "dry_run": true}'

No API, no scheduling — just run the job once with your own parameters.

Programmatic use¶

The CLI is a thin wrapper around JobRunner.run(), which you can call directly:

import asyncio
from craft_easy_jobs.runner import JobRunner

async def main():
    result = await JobRunner.run(
        "settlement",
        params={"period": "2026-03"},
        tenant_id="tenant_664abc",
    )
    print(result.status)              # "completed" or "failed"
    print(result.duration_seconds)
    print(result.result)              # Return value
    for log in result.logs:
        print(log["level"], log["message"])

asyncio.run(main())

JobResult exposes everything the CLI prints, so embedding the runner in your own tooling is straightforward.

Built-in jobs¶

Three jobs are shipped with the package and become available when you install craft-easy-jobs[full]:

Job	Description	Timeout	Retries
`token-cleanup`	Delete expired `UserAuthentication` sessions	120s	0
`audit-log-archive`	Archive old audit log entries	600s	0
`bi-export`	Export resources to BigQuery or Azure SQL	3600s	2

See Built-in Jobs for parameter reference and usage examples.

FileImportQueue¶

craft-easy-jobs ships a MongoDB-backed priority queue purpose-built for file import workloads. It lets any number of workers compete for jobs without a central coordinator — MongoDB provides the serialisation.

Creating the queue¶

from motor.motor_asyncio import AsyncIOMotorClient
from craft_easy_jobs.queues import FileImportQueue

client = AsyncIOMotorClient(settings.MONGODB_URI)
db = client[settings.DATABASE_NAME]

queue = FileImportQueue(db=db)

If db is omitted and craft-easy-api is installed, the queue auto-connects using the application's MONGODB_URI and DATABASE_NAME settings. In that case, construct the queue with no arguments:

queue = FileImportQueue()   # auto-connects via craft_easy.settings

Enqueuing jobs¶

job_id = await queue.enqueue(
    run_id="import_run_abc123",
    template_id="template_invoice",
    source_id="sftp_source_001",
    file_path="/uploads/january_invoices.csv",
    priority=8,
    max_runtime_minutes=60,
    max_memory_mb=2048,
    params={"tenant_id": "tenant_xyz"},
)

Parameter	Default	Purpose
`run_id`	required	Import run identifier forwarded to the processing job
`template_id`	required	Template that describes how to parse the file
`source_id`	`""`	Origin of the file (e.g. an SFTP source ID)
`file_path`	`""`	Path to the file to import
`priority`	`5`	Integer priority — higher values are dequeued first
`max_runtime_minutes`	`240`	Per-job hard timeout
`max_memory_mb`	`4096`	Memory limit enforced by the worker process
`params`	`{}`	Arbitrary parameters forwarded to the processing job

Priority mechanism¶

Jobs are sorted by (priority DESC, enqueued_at ASC). Within the same priority level, jobs are processed in strict FIFO order. Acquisition is a single atomic find_one_and_update with that sort, so no two workers can claim the same job.

Priority 10 ──── dequeued first; FIFO within priority 10
Priority 5  ──── standard queue; FIFO within priority 5
Priority 1  ──── background / bulk imports

acquire() returns None when the queue is empty; the worker then sleeps for poll_interval_seconds (default 5 s) before trying again.

Leases and heartbeats¶

When a worker acquires a job, the queue records lease_expires_at = now + 900 s. The worker must renew the lease before it expires by calling heartbeat(). FileImportWorker does this automatically every 60 s:

signal = await queue.heartbeat(job_id)   # returns any pending signal, e.g. "cancel"

If a worker crashes without releasing a job, the lease expires naturally and the job becomes eligible for re-acquisition. Workers run the reaper on startup to recover any stale leases immediately:

count = await queue.reap_expired_leases()   # returns number of jobs reclaimed

Dead letter handling¶

Transient errors (network timeouts, 5xx responses, connection failures) are retried automatically with exponential backoff:

Attempt	Wait before next attempt
1 → 2	30 s
2 → 3	120 s
3 → 4	300 s
4 (final)	Permanent failure → DLQ

After the final attempt the job document is copied to file_import_jobs_dlq with dlq_reason and moved_at fields set. Non-transient errors (parse failures, OOM, explicit timeout) skip retries and go directly to failed status without entering the DLQ.

# Inspect dead-letter jobs
dlq = db["file_import_jobs_dlq"]
async for doc in dlq.find({}):
    print(doc["_id"], doc["dlq_reason"], doc["moved_at"])

MongoDB collections¶

Collection	Purpose
`file_import_jobs`	Active queue: `pending`, `running`, `retry_pending`, `completed`, `failed`, `cancelled`
`file_import_jobs_dlq`	Dead letter: permanently failed jobs copied here for manual review

Horizontal scaling with FileImportWorker¶

FileImportWorker is designed to scale horizontally from day one. Each instance runs an independent poll loop — there is no primary/replica distinction and no external coordination service needed.

How it works¶

Worker calls queue.acquire() — MongoDB atomically marks the job running and records the worker ID and lease expiry.
Worker runs file_import.process_file via JobRunner with per-job resource limits.
Worker calls queue.complete() or queue.fail() when the job finishes.
Worker loops back to step 1.

Because find_one_and_update is atomic, any number of workers can point at the same MongoDB and each job is processed by exactly one worker.

Running multiple workers¶

import asyncio, os
from motor.motor_asyncio import AsyncIOMotorClient
from craft_easy_jobs.queues import FileImportQueue
from craft_easy_jobs.workers import FileImportWorker

async def start_worker(worker_id: str, db):
    queue = FileImportQueue(db=db)
    worker = FileImportWorker(
        queue=queue,
        worker_id=worker_id,
        api_base_url=os.environ["API_BASE_URL"],
        api_token=os.environ["WORKER_API_TOKEN"],
    )
    await worker.run()

async def main():
    client = AsyncIOMotorClient(os.environ["MONGODB_URI"])
    db = client["myapp"]
    await asyncio.gather(
        start_worker("worker-0", db),
        start_worker("worker-1", db),
        start_worker("worker-2", db),
    )

asyncio.run(main())

In production, each worker is a separate process or container so resource limits (memory, CPU) are enforced independently.

Worker configuration reference¶

Parameter	Default	Purpose
`worker_id`	`worker-{pid}`	Identifies the worker in logs and on the job document
`poll_interval_seconds`	`5`	How long to sleep when the queue is empty
`heartbeat_interval_seconds`	`60`	How often to renew the job lease during execution
`api_base_url`	`http://localhost:5001`	API endpoint the processing job calls back to
`api_token`	`""`	Service token used for authenticated API callbacks

Deployment configurations¶

Kubernetes Deployment — recommended for persistent, high-throughput workloads:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: file-import-worker
spec:
  replicas: 3                         # scale up/down based on queue depth
  selector:
    matchLabels:
      app: file-import-worker
  template:
    metadata:
      labels:
        app: file-import-worker
    spec:
      containers:
      - name: worker
        image: my-registry/craft-easy-jobs:latest
        command: ["python", "-m", "my_app.workers.file_import"]
        env:
        - name: MONGODB_URI
          valueFrom:
            secretKeyRef:
              name: craft-easy-secrets
              key: mongodb-uri
        - name: WORKER_API_TOKEN
          valueFrom:
            secretKeyRef:
              name: craft-easy-secrets
              key: worker-api-token
        resources:
          requests:
            memory: "4Gi"
            cpu: "500m"
          limits:
            memory: "8Gi"

Adding a replica takes effect immediately — there is no rebalancing or configuration reload required.

Docker Compose — for development and staging:

services:
  file-import-worker:
    image: my-registry/craft-easy-jobs:latest
    command: python -m my_app.workers.file_import
    deploy:
      replicas: 2
    environment:
      MONGODB_URI: mongodb://mongo:27017
      WORKER_API_TOKEN: ${WORKER_API_TOKEN}
    depends_on:
      - mongo

Google Cloud Run Jobs — for burst workloads triggered by Cloud Scheduler:

gcloud run jobs create file-import-worker \
  --image my-registry/craft-easy-jobs:latest \
  --tasks 5 \
  --parallelism 5 \
  --max-retries 0 \
  --set-env-vars MONGODB_URI=... \
  --command python,-m,my_app.workers.file_import

Set --max-retries 0 — job-level retries are handled by FileImportQueue, not by Cloud Run. Each task maps to one worker process, and all five tasks compete for jobs from the shared queue.

Limits of the standalone runner¶

The CLI is intentionally minimal. Things it does not do:

No persistent history. JobResult is returned to the caller and that's it. If you need to keep it, write it to your own store.
No scheduling. Use cron, Cloud Scheduler, GitHub Actions, or Kubernetes CronJobs to trigger it.
No distributed locking. If you invoke the same CLI twice concurrently, you will get two concurrent runs. For single-execution guarantees, use the API framework or a trigger system that dedupes.
No REST API. The CLI is the interface.
No chain execution. The chain() helper lets you define chains, but triggering them requires the API framework (which reads the chain off a JobSchedule).

If any of these are dealbreakers for your use case, use the API job framework instead.