CSV Import Pipeline

I wanted Pocket Family deployable on two stacks, self-hosted VPS and AWS, so the import logic itself couldn’t know or care which one it was running on. Everything stack-specific goes through environment variables read by one Settings class (import-service/app/config.py): broker_url is redis://... locally and sqs:// on AWS, storage_backend switches between local and s3. The same Docker image runs on both stacks with the same code, just a different env file, so switching stacks means changing config, not rewriting code.

One Core, Two Triggers

The import logic is process_import() in import_csv.py, and it doesn’t import Celery. Two thin wrappers call into it:

• celery_tasks.py — the local/self-hosted path. A Celery task (import_service.execute_import) that delegates straight to process_import(), running on Redis as the broker.
• lambda_handler.py — the AWS path. No Celery worker runs on AWS; a Lambda is wired directly to the SQS queue and imports process_import itself.

The backend dispatches the same way on both stacks, via Celery’s send_task(...). Locally that lands on a Celery worker. On AWS, Celery/kombu is configured to publish straight to SQS instead, and the Lambda image is built without celery/kombu/redis installed, so it can only import the celery-free core.

Envelope Decoding

Because the message on the wire is a Celery/kombu-formatted envelope, not raw JSON, the Lambda has to unwrap it before it can call process_import():

1. The raw SQS body is JSON, unless kombu base64-encoded the whole thing, so it tries plain json.loads first and falls back to base64.
2. That gives an envelope dict with the real message in envelope["body"], itself base64-encoded again if envelope["properties"]["body_encoding"] == "base64".
3. The decoded inner body is a Celery protocol v2 message, [args, kwargs, embed]. The backend dispatches with kwargs={"payload": {...}}, so the payload is kwargs["payload"].

That’s three layers of decoding just to recover one dict. It’s isolated in a single function (decode_sqs_record) and unit-tested on its own, since it couples the Lambda to Celery’s wire format.

Idempotency

process_import() claims its work with one conditional update, run as the first statement:

UPDATE importjob SET status = 'STARTED', updated_at = :now
WHERE id = :id AND status != 'DONE'

If rowcount == 0, the job was already finished and the function returns {"status": "skipped"} without touching any data. That’s what makes retries safe on both stacks: Celery’s autoretry_for on the local path, SQS’s redelivery on AWS. A message replayed after a crash just re-claims and no-ops instead of double-importing.