Using Celery for Async Flight Schedule Batches

The exact problem: validate schedule deltas off the request thread

Flight operations managers and crew schedulers routinely process schedule deltas that arrive as fragmented CSV, XML, or JSON payloads from an operations control centre. Each payload carries flight segments, block times, and assigned crew identifiers that must be checked against cumulative duty limits, mandatory rest, and time-zone transition rules before they are committed to the crew management system. When that check runs synchronously inside the ingestion endpoint, database row locks and external regulatory API calls spike request latency, block roster publication, and raise the risk of a non-compliant roster going live.

The concrete task this page solves: accept a batch of schedule segments quickly at the ingestion boundary, hand the heavy compliance validation to an isolated Celery worker, and return a structured, reproducible verdict against FAA Part 117 rule schema and EASA FTL compliance thresholds — without ever blocking the endpoint and without double-committing a crew assignment on retry. This how-to sits within the async batch processing workflows area of the wider flight data ingestion architecture, and assumes each segment’s report and block times have already been normalised to UTC and classified against the shared crew duty time taxonomy upstream.

Prerequisites checklist

Confirm the following before wiring up the worker:

Step-by-step implementation

Step 1 — Configure the Celery app for at-least-once, isolated execution

Start with a broker and worker configuration tuned for large, order-sensitive schedule payloads. task_acks_late=True guarantees a task is only acknowledged after it completes, so a crashed worker re-delivers rather than drops the batch; worker_prefetch_multiplier=1 stops a single worker from hoarding the queue and starving its peers during a heavy ingestion window.

from celery import Celery

app = Celery("flight_schedule_validator")
app.conf.update(
    broker_connection_retry_on_startup=True,
    worker_prefetch_multiplier=1,      # one heavy batch per worker at a time
    task_acks_late=True,               # ack only after successful completion
    task_reject_on_worker_lost=True,   # re-queue if the worker dies mid-task
    task_serializer="json",
    result_serializer="json",
    task_routes={
        # High-priority duty-limit checks jump ahead of routine roster edits.
        "tasks.validate_schedule_batch": {"queue": "compliance_critical"},
    },
)

Verifiable output: starting a worker with celery -A tasks worker -Q compliance_critical --concurrency=4 and inspecting celery -A tasks inspect active_queues confirms the worker is bound to the critical queue and prefetching one message at a time. Routing keys let flight ops prioritise urgent duty-limit checks over administrative roster adjustments, so a delay-driven duty extension is never stuck behind a batch of cosmetic edits.

Step 2 — Define a typed segment model at the task boundary

Wrap every incoming segment in a Pydantic model so a malformed payload fails at the boundary rather than deep inside the evaluator. Store the full Flight Duty Period explicitly — fdp_minutes is not block time, and conflating the two is the most common source of false verdicts.

from pydantic import BaseModel


class ScheduleSegment(BaseModel):
    crew_id: str
    flight_number: str
    departure_utc: str          # ISO 8601, timezone-aware UTC
    arrival_utc: str            # ISO 8601, timezone-aware UTC
    # fdp_minutes: total Flight Duty Period in minutes (report time -> block-on),
    # supplied by the upstream roster system and stored separately from block time.
    # FDP begins at the official report time and ends at final block-on, encompassing
    # all pre-flight and post-flight duty; FDP != block time.
    fdp_minutes: int

Verifiable output: ScheduleSegment.model_validate({...}) on a well-formed segment returns a typed instance, while a missing fdp_minutes raises ValidationError — proving the boundary rejects incomplete records before any compliance logic runs.

Step 3 — Implement the batch task with idempotent retry

The core task normalises each segment, applies the duty-limit rule, and returns a deterministic report. bind=True exposes self.request.retries so transient database or external-API failures retry with exponential backoff, while a per-segment ValidationError is quarantined without aborting the whole batch.

import logging
from typing import Any

from pydantic import ValidationError

logger = logging.getLogger(__name__)


@app.task(bind=True, max_retries=3, default_retry_delay=60, acks_late=True)
def validate_schedule_batch(
    self, batch_id: str, segments: list[dict[str, Any]]
) -> dict[str, Any]:
    """Validate a batch of segments against FAR 117 / EASA FTL thresholds."""
    try:
        validated: list[dict[str, Any]] = []
        violations: list[dict[str, Any]] = []

        for seg_data in segments:
            try:
                segment = ScheduleSegment(**seg_data)
            except ValidationError as exc:
                logger.warning("Schema validation failed for segment: %s", exc)
                continue

            is_compliant, reason = _evaluate_duty_limits(segment)
            if is_compliant:
                validated.append(segment.model_dump())
            else:
                violations.append({
                    "crew_id": segment.crew_id,
                    "flight_number": segment.flight_number,
                    "violation": reason,
                })

        return {
            "batch_id": batch_id,
            "status": "COMPLETED",
            "total_processed": len(segments),
            "compliant_count": len(validated),
            "violation_count": len(violations),
            "violations": violations,
        }
    except Exception as exc:  # transient infra failure -> retry with backoff
        logger.error("Batch %s failed during validation: %s", batch_id, exc)
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

Verifiable output: enqueue with validate_schedule_batch.delay("B-001", segments) and fetch the AsyncResult; a mixed batch returns status="COMPLETED" with compliant_count + violation_count accounting for every well-formed segment, and the retry countdown doubles (1s, 2s, 4s) across successive infrastructure failures.

Step 4 — Encode the duty-limit rule against the §117.13 ceiling

Keep the regulatory decision in one deterministic function so the constant is auditable and versionable. FAR §117.13 Table B sets the maximum FDP for a two-pilot crew between 9 and 14 hours depending on the acclimated report time and segment count; the 14-hour value is the absolute ceiling that applies only under the most favourable conditions (report 0800–1159 local, one segment). A production engine looks up the applicable row and column; the stub below enforces only the absolute cap.

def _evaluate_duty_limits(segment: ScheduleSegment) -> tuple[bool, str]:
    """Deterministic §117.13 ceiling check.

    Replaced in production by a cached, context-aware rule engine that also
    evaluates rolling 24h/48h/7d cumulative FDP (§117.23), minimum rest
    (§117.25), and acclimatisation adjustments. fdp_minutes is the full Flight
    Duty Period, not block time.
    """
    if segment.fdp_minutes > 840:   # 14 h x 60 = 840 min, absolute §117.13 ceiling
        return False, "FDP exceeds absolute 14-hour ceiling per FAR §117.13 Table B"
    return True, ""

Verifiable output: a segment with fdp_minutes=900 returns (False, "FDP exceeds ...") and surfaces in the batch violations list, while fdp_minutes=780 returns (True, "") — pinning the exact 840-minute boundary the regulation draws.

The end-to-end task flow — endpoint to broker to isolated workers to result backend, with acks_late retries on failure — is shown below.

Celery async schedule-batch task flow The ingestion endpoint enqueues a validation task with .delay() and returns immediately. The task lands on a durable broker (Redis or RabbitMQ) in the compliance_critical queue. Isolated workers pull one batch each (prefetch = 1) and evaluate every segment against Flight Duty Period, rest, and cumulative limits. A COMPLETED verdict is persisted to the result backend keyed on batch id, payload hash, and rule version. If a worker crashes or a transient failure is raised, acks_late re-delivers the unacknowledged batch back to the broker instead of dropping it. .delay() prefetch=1 COMPLETED failure → acks_late re-deliver Ingestion endpoint accept batch → 202 Broker Redis / RabbitMQ queue: compliance_critical Worker 1 FDP · rest · cumulative Worker N FDP · rest · cumulative ⋮ Result backend structured report batch_id · rule ver

Figure: Celery task flow: the endpoint enqueues to a broker, isolated workers run regulatory validation, and results persist to the backend; failures retry under acks_late.

Verification queries and assertions

Pin the task’s behaviour with pytest so a future refactor cannot quietly move the ceiling or break the quarantine path. Run the task eagerly in-process to assert on the returned report without a live broker.

def test_batch_flags_ceiling_breach():
    segments = [
        {"crew_id": "CM-1", "flight_number": "AB100",
         "departure_utc": "2026-07-01T08:00:00+00:00",
         "arrival_utc": "2026-07-01T20:00:00+00:00", "fdp_minutes": 900},
        {"crew_id": "CM-2", "flight_number": "AB101",
         "departure_utc": "2026-07-01T08:00:00+00:00",
         "arrival_utc": "2026-07-01T16:00:00+00:00", "fdp_minutes": 780},
    ]
    result = validate_schedule_batch.apply(args=("B-TEST", segments)).get()
    assert result["status"] == "COMPLETED"
    assert result["violation_count"] == 1
    assert result["compliant_count"] == 1
    assert result["violations"][0]["crew_id"] == "CM-1"


def test_malformed_segment_is_quarantined_not_fatal():
    segments = [{"crew_id": "CM-3", "flight_number": "AB102"}]  # missing fields
    result = validate_schedule_batch.apply(args=("B-BAD", segments)).get()
    assert result["status"] == "COMPLETED"
    assert result["total_processed"] == 1
    assert result["compliant_count"] == 0 and result["violation_count"] == 0

Set app.conf.task_always_eager = True in the test configuration so .apply().get() runs the task synchronously; two passing assertions confirm the ceiling check fires and a malformed segment is skipped rather than crashing the batch.

Failure modes and troubleshooting

Frequently asked questions

Should I use Redis or RabbitMQ as the broker?

Both work; the trade-off is operational. Redis is lighter to run and doubles as the result backend, which suits a single-region deployment evaluating duty limits. RabbitMQ gives stronger delivery semantics, priority queues, and per-message TTL, which matter when a dual-jurisdiction carrier needs strict ordering and priority routing between compliance_critical and routine queues. Whichever you choose, enable persistence so queued validations survive a restart.

How do I guarantee a retried batch does not double-commit a crew assignment?

Combine acks_late=True with an idempotent commit. Give each batch_job an idempotency key and make the database write an upsert on that key rather than an insert, so a re-delivered task claims the job once and every subsequent attempt is a no-op. The task’s compute path can be re-run safely; only the commit needs to be guarded.

Does the 840-minute ceiling cover the whole of Part 117?

No. The 840-minute figure is only the absolute §117.13 Table B ceiling. A production check must look up the applicable Table B row and column for the acclimated report time and segment count (which can cap the FDP as low as 9 hours), then evaluate the §117.23 rolling cumulative windows (100 h / 672 h, 60 h / 168 h, and the annual caps) and the §117.25 rest minimum. The stub isolates the simplest rule so the pattern is clear; the real engine is cached and context-aware.

How does a schedule delta that crosses a DST transition stay correct?

Keep every instant in UTC through the task and derive the local report hour that keys the Table B band only at evaluation, using the segment’s originating zone. Because departure_utc and fdp_minutes are absolute, the FDP duration is unaffected by a wall-clock change; only the local-hour lookup needs the zone. Reject naive timestamps at ingestion so a DST boundary never silently shifts the band.

Where do violations go once the worker flags them?

Into the structured result backend, and — for hard breaches — onto an event the scheduler consumes asynchronously, never a blocking call back into the ingestion endpoint. Soft flags (a non-preferred rest facility) can be reviewed and overridden; hard §117.13 or cumulative breaches are enforced. Every verdict is persisted with the batch ID, payload hash, and rule version behind the system security and access boundaries controls, so an FAA or EASA audit can replay exactly how a roster was cleared.

Back to Async Batch Processing Workflows.