How to Build an Image Processing Pipeline with S3

01

What we are building

We are building an event-driven image processing pipeline for new S3 uploads. A user or app uploads an image into a private input bucket or prefix. S3 sends an ObjectCreated event to SQS. A Lambda worker pulls one message at a time, creates or updates a DynamoDB job record, generates a short-lived presigned URL for the source object, calls BackgroundErase, validates the image response, and writes the processed file to an output bucket or prefix.

input image upload
-> s3://my-image-pipeline-input/input/product-1.jpg
-> S3 ObjectCreated event
-> SQS job queue
-> Lambda worker
-> DynamoDB job record and rate-limit slot
-> BackgroundErase API
-> s3://my-image-pipeline-output/output/product-1.png

The input bucket stays private. The API receives a presigned URL that expires quickly, not public bucket access and not a forever URL you shoved into a database six hours before a worker finally got around to it. Store bucket, key, and version ID; generate the URL right before the API call.

02

Why not call the API directly from S3 events?

The tempting version is very short. It is also exactly the version I would avoid for a production upload path unless the volume is tiny and you are comfortable babysitting it.

S3 ObjectCreated
-> Lambda
-> BackgroundErase API
-> output S3

Direct S3 to Lambda is fine for a demo. It is not the architecture I would ship for a pipeline that may receive thousands or millions of images at once. Treat S3, SQS, and Lambda as at-least-once systems and design the worker so a duplicate event does not mean a duplicate API charge.

03

Production architecture

The production architecture adds two boring-looking pieces that do a lot of work: SQS and DynamoDB. SQS absorbs upload bursts and gives Lambda a backlog to drain. DynamoDB records durable job state, prevents duplicate processing, and holds the global rate-limit clock.

S3 input bucket:
  private raw uploads

SQS Standard queue:
  burst buffer and retry surface

Lambda worker:
  small processor that pulls one image job at a time

DynamoDB jobs table:
  idempotency, status, audit history, output pointers

DynamoDB rate-limit table:
  one global slot clock for the BackgroundErase API quota

S3 output bucket:
  processed PNGs, JPEGs, WebPs, or downstream-ready assets

CloudWatch:
  backlog age, failures, duration, throttles, DLQ alarms

Dead-letter queue:
  messages that failed too many times and need inspection

DynamoDB still matters a lot. It is the difference between "some Lambdas ran" and "this exact input object version became this output object at this time, after this many attempts, with this final status." That audit trail becomes extremely nice the first time a customer asks where 138 missing images went.

04

Create the S3 buckets

Use either separate buckets or strict prefixes. Separate buckets are cleaner operationally. One bucket with input/ and output/ prefixes is also fine if the notification filter is tight. The one thing you do not want is output uploads accidentally triggering the same pipeline again. Recursive image processing is funny for about eight seconds.

# Two buckets
s3://backgrounderase-pipeline-input/input/product-1.jpg
s3://backgrounderase-pipeline-output/output/product-1.png

# Or one bucket with strict prefixes
s3://my-bucket/input/product-1.jpg
s3://my-bucket/output/product-1.png

Versioning matters because uploads can race with processing. If a user overwrites input/product-1.jpg while the queue is backed up, the worker should not accidentally process the newer file for the older event. Include versionId in the job key whenever it is available.

05

Add the SQS queue and dead-letter queue

Create the dead-letter queue first, then create the main queue. The main queue is the one that receives S3 ObjectCreated events. The DLQ is only the destination for messages that fail too many times in the main queue. This ordering matters in the AWS Console because you usually select the DLQ while configuring the main queue's dead-letter queue settings.

Dead-letter queue:
  name: backgrounderase-image-jobs-dlq
  type: Standard
  message retention: 14 days
  encryption: SSE-SQS or SSE-KMS

Main queue:
  name: backgrounderase-image-jobs
  type: Standard
  visibility timeout: 5-10 minutes
  message retention: 14 days
  receive message wait time: 20 seconds
  dead-letter queue: backgrounderase-image-jobs-dlq
  maximum receives / maxReceiveCount: 5

If you are using the AWS Console, create backgrounderase-image-jobs-dlq from SQS first. Then create backgrounderase-image-jobs, enable the dead-letter queue option, choose the DLQ you just made, and set Maximum receives to 5. If you prefer the CLI, run the block below. It stores the returned queue URLs and ARNs in variables and uses them later in the same script, so you do not have to paste <dead-letter-queue-url> into another command. The CLI snippets in this article run inside a child Bash process, so a failed command stops that snippet without closing your current terminal session.

bash <<'BASH'
set -euo pipefail

DLQ_NAME=backgrounderase-image-jobs-dlq
MAIN_QUEUE_NAME=backgrounderase-image-jobs

DLQ_URL=$(aws sqs create-queue \
  --queue-name "$DLQ_NAME" \
  --attributes MessageRetentionPeriod=1209600,SqsManagedSseEnabled=true \
  --query QueueUrl \
  --output text)

DLQ_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$DLQ_URL" \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' \
  --output text)

cat > main-queue-attributes.json <<JSON
{
  "VisibilityTimeout": "600",
  "MessageRetentionPeriod": "1209600",
  "ReceiveMessageWaitTimeSeconds": "20",
  "SqsManagedSseEnabled": "true",
  "RedrivePolicy": "{\"deadLetterTargetArn\":\"$DLQ_ARN\",\"maxReceiveCount\":\"5\"}"
}
JSON

MAIN_QUEUE_URL=$(aws sqs create-queue \
  --queue-name "$MAIN_QUEUE_NAME" \
  --attributes file://main-queue-attributes.json \
  --query QueueUrl \
  --output text)

MAIN_QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' \
  --output text)

printf 'Main queue URL: %s\nMain queue ARN: %s\nDLQ URL: %s\nDLQ ARN: %s\n' \
  "$MAIN_QUEUE_URL" "$MAIN_QUEUE_ARN" "$DLQ_URL" "$DLQ_ARN"
BASH

The DLQ does not need its own redrive policy. It is the parking lot for messages that the worker repeatedly failed to process. The redrive policy belongs to the main queue, and it points at the DLQ. If your main queue is Standard, use a Standard DLQ. If your main queue is FIFO, use a FIFO DLQ. For this article, both queues should be Standard.

In the AWS Console, you may not see a field literally called maxReceiveCount. That is the API and CloudFormation name. In the Console, it is usually shown under the dead-letter queue or redrive policy settings as Maximum receives. Set it to 5. That means SQS can deliver the same message to Lambda and see it fail up to five times before moving the message from backgrounderase-image-jobs to backgrounderase-image-jobs-dlq.

A value of five is a reasonable starting point for this pipeline: enough room for a transient timeout or rate-limit bump, not so much room that one corrupt image clogs the queue forever. Messages in the DLQ are not automatically fixed. Inspect the DynamoDB job record, decide whether the input is replayable, then either replay the message or mark the job permanently failed.

On the main queue, use long polling so Lambda is not constantly asking an empty queue if it has changed its mind. Set the visibility timeout longer than normal processing time, otherwise one slow job can become two workers doing the same work and acting surprised about it.

Next, let S3 send messages to the main queue. This policy belongs on backgrounderase-image-jobs, not on the DLQ and not on the S3 bucket. In the AWS Console, open SQS, choose the main queue, edit its Access policy, and grant s3.amazonaws.com permission to call sqs:SendMessage from your input bucket. The CLI block below writes the policy with your current AWS account ID and queue ARN, then applies it to the main queue.

bash <<'BASH'
set -euo pipefail

INPUT_BUCKET=backgrounderase-pipeline-input
MAIN_QUEUE_NAME=backgrounderase-image-jobs

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
MAIN_QUEUE_URL=$(aws sqs get-queue-url \
  --queue-name "$MAIN_QUEUE_NAME" \
  --query QueueUrl \
  --output text)
MAIN_QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' \
  --output text)

cat > sqs-policy.json <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowS3ToSendObjectCreatedEvents",
      "Effect": "Allow",
      "Principal": {
        "Service": "s3.amazonaws.com"
      },
      "Action": "sqs:SendMessage",
      "Resource": "$MAIN_QUEUE_ARN",
      "Condition": {
        "ArnEquals": {
          "aws:SourceArn": "arn:aws:s3:::$INPUT_BUCKET"
        },
        "StringEquals": {
          "aws:SourceAccount": "$ACCOUNT_ID"
        }
      }
    }
  ]
}
JSON

python3 - <<'PY'
import json

with open("sqs-policy.json") as src:
    policy = json.load(src)

with open("sqs-policy-attributes.json", "w") as dst:
    json.dump({"Policy": json.dumps(policy)}, dst)
PY

aws sqs set-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attributes file://sqs-policy-attributes.json
BASH

After the queue can accept S3 events, configure the input bucket to send ObjectCreated events into the main queue. In the AWS Console, open the input bucket, go to Properties - Event notifications, create a new notification, set the prefix to input/, choose all object-created events, choose SQS as the destination, and select backgrounderase-image-jobs. From the CLI, this block writes the bucket notification file with the correct queue ARN and applies it to backgrounderase-pipeline-input.

bash <<'BASH'
set -euo pipefail

INPUT_BUCKET=backgrounderase-pipeline-input
MAIN_QUEUE_NAME=backgrounderase-image-jobs

MAIN_QUEUE_URL=$(aws sqs get-queue-url \
  --queue-name "$MAIN_QUEUE_NAME" \
  --query QueueUrl \
  --output text)
MAIN_QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' \
  --output text)

cat > notification.json <<JSON
{
  "QueueConfigurations": [
    {
      "Id": "send-input-images-to-backgrounderase-queue",
      "QueueArn": "$MAIN_QUEUE_ARN",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {
              "Name": "prefix",
              "Value": "input/"
            }
          ]
        }
      }
    }
  ]
}
JSON

aws s3api put-bucket-notification-configuration \
  --bucket "$INPUT_BUCKET" \
  --notification-configuration file://notification.json
BASH

Before moving on, test the wiring while Lambda is still disconnected. Upload one harmless sample image to the input prefix and check the main queue attributes. You should see the visible message count move above zero. If it does not, fix the queue policy or bucket notification before adding a worker.

bash <<'BASH'
set -euo pipefail

INPUT_BUCKET=backgrounderase-pipeline-input
MAIN_QUEUE_NAME=backgrounderase-image-jobs
MAIN_QUEUE_URL=$(aws sqs get-queue-url \
  --queue-name "$MAIN_QUEUE_NAME" \
  --query QueueUrl \
  --output text)

aws s3 cp ./sample.jpg "s3://$INPUT_BUCKET/input/sample.jpg"

aws sqs get-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible
BASH

This notification only filters by prefix. You can add suffix rules for .jpg, .jpeg, .png, and .webp, but that usually means multiple notification rules. For this pipeline, sending every input/ object to the queue and letting the Lambda worker mark unsupported files as SKIPPED is simpler and harder to misconfigure.

06

Create the DynamoDB tables

The jobs table is the durable ledger. Do not delete successful jobs immediately. Mark them COMPLETED, store output pointers, and use TTL to expire old records later. Deleting successful jobs makes audits worse and makes duplicate events more expensive.

Create both tables with on-demand billing. The jobs table stores one row per input object version. The rate-limit table stores one small shared row used by every Lambda worker before calling the API. The script waits for each table before enabling TTL or writing the initial limiter row; DynamoDB often returns from create-table while the table is still CREATING.

bash <<'BASH'
set -euo pipefail

JOBS_TABLE=BackgroundEraseJobs
RATE_LIMIT_TABLE=BackgroundEraseRateLimit

if ! aws dynamodb describe-table --table-name "$JOBS_TABLE" >/dev/null 2>&1; then
  aws dynamodb create-table \
    --table-name "$JOBS_TABLE" \
    --attribute-definitions AttributeName=PK,AttributeType=S \
    --key-schema AttributeName=PK,KeyType=HASH \
    --billing-mode PAY_PER_REQUEST
fi

aws dynamodb wait table-exists --table-name "$JOBS_TABLE"

TTL_STATUS=$(aws dynamodb describe-time-to-live \
  --table-name "$JOBS_TABLE" \
  --query 'TimeToLiveDescription.TimeToLiveStatus' \
  --output text 2>/dev/null || true)

if [ "$TTL_STATUS" != "ENABLED" ] && [ "$TTL_STATUS" != "ENABLING" ]; then
  aws dynamodb update-time-to-live \
    --table-name "$JOBS_TABLE" \
    --time-to-live-specification "Enabled=true, AttributeName=ttl"
fi

if ! aws dynamodb describe-table --table-name "$RATE_LIMIT_TABLE" >/dev/null 2>&1; then
  aws dynamodb create-table \
    --table-name "$RATE_LIMIT_TABLE" \
    --attribute-definitions AttributeName=PK,AttributeType=S \
    --key-schema AttributeName=PK,KeyType=HASH \
    --billing-mode PAY_PER_REQUEST
fi

aws dynamodb wait table-exists --table-name "$RATE_LIMIT_TABLE"

aws dynamodb put-item \
  --table-name "$RATE_LIMIT_TABLE" \
  --item '{"PK":{"S":"RATE_LIMIT#backgrounderase"},"next_allowed_at_ms":{"N":"0"}}'
BASH

Table: BackgroundEraseJobs

Primary key:
  PK = JOB#{bucket}#{key}#{versionId}

Attributes:
  status              PENDING | IN_PROGRESS | COMPLETED | FAILED | SKIPPED
  input_bucket
  input_key
  input_version_id
  input_etag
  input_size
  output_bucket
  output_key
  attempt_count
  last_error
  created_at
  updated_at
  lease_expires_at
  completed_at
  ttl

On every message, the worker attempts a conditional PutItem with attribute_not_exists(PK). If the item already exists and the status is COMPLETED, return success without calling the API again. If the status is IN_PROGRESS but lease_expires_at is in the past, a worker can reclaim the job. If it is FAILED, decide whether the failure was permanent or worth another attempt.

Table: BackgroundEraseRateLimit

Primary key:
  PK = RATE_LIMIT#backgrounderase

Attributes:
  next_allowed_at_ms

The put-item command seeds the rate limiter with next_allowed_at_ms=0. Without that row, the worker can still create it with a slightly different implementation, but seeding it up front makes the first run boring, which is the goal.

07

Configure the Lambda worker

The worker should be intentionally small and boring. It does not need to download the image unless you are doing local validation. It needs to parse the S3 event, write job state, generate a presigned URL, call the API, validate the response, upload the output, and update the job record.

Use SQS batch size 1. The API is the bottleneck, not Lambda throughput. A batch of ten makes partial failure handling more important and makes "which image caused this retry?" more annoying. One message per invocation is boring in the best possible way.

Create an IAM role for the function first, then create the Lambda function from lambda_worker.py. The CLI block below creates the Lambda trust policy, creates the role if it does not already exist, writes the inline permissions policy with your queue and account values, and attaches it to the role. In the Console, the same Lambda settings are under Configuration: timeout, memory, reserved concurrency, environment variables, and triggers. The trigger should be the main SQS queue, not the DLQ.

INPUT_BUCKET=backgrounderase-pipeline-input
INPUT_PREFIX=input/
OUTPUT_BUCKET=backgrounderase-pipeline-output
OUTPUT_PREFIX=output/
JOBS_TABLE=BackgroundEraseJobs
RATE_LIMIT_TABLE=BackgroundEraseRateLimit
BACKGROUNDERASE_SECRET_NAME=backgrounderase/api-key
BACKGROUNDERASE_API_URL=https://api.backgrounderase.com/v2
PRESIGNED_URL_EXPIRATION_SECONDS=900
MAX_INPUT_FILE_SIZE_BYTES=30000000

import json
import os
import random
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import PurePosixPath

import boto3
from botocore.exceptions import ClientError


s3 = boto3.client("s3")
ddb = boto3.resource("dynamodb")
ddb_raw = boto3.client("dynamodb")
secrets = boto3.client("secretsmanager")

INPUT_BUCKET = os.environ["INPUT_BUCKET"]
INPUT_PREFIX = os.getenv("INPUT_PREFIX", "input/")
OUTPUT_BUCKET = os.environ["OUTPUT_BUCKET"]
OUTPUT_PREFIX = os.getenv("OUTPUT_PREFIX", "output/")
JOBS_TABLE = os.environ["JOBS_TABLE"]
RATE_LIMIT_TABLE = os.environ["RATE_LIMIT_TABLE"]
SECRET_NAME = os.environ["BACKGROUNDERASE_SECRET_NAME"]
API_URL = os.getenv("BACKGROUNDERASE_API_URL", "https://api.backgrounderase.com/v2")
PRESIGN_SECONDS = int(os.getenv("PRESIGNED_URL_EXPIRATION_SECONDS", "900"))
MAX_BYTES = int(os.getenv("MAX_INPUT_FILE_SIZE_BYTES", "30000000"))

SUPPORTED_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"}
LEASE_SECONDS = 8 * 60
JOB_TTL_SECONDS = 45 * 24 * 60 * 60
RATE_PK = "RATE_LIMIT#backgrounderase"

jobs = ddb.Table(JOBS_TABLE)
be_key_cache = None


class BadSource(Exception):
    pass


class TryAgain(Exception):
    pass


def lambda_handler(event, context):
    for msg in event.get("Records", []):
        handle_sqs_msg(msg)
    return {"ok": True}


def handle_sqs_msg(msg):
    body = json.loads(msg.get("body") or "{}")

    if body.get("Event") == "s3:TestEvent":
        return

    for rec in body.get("Records", []):
        handle_s3_rec(rec)


def handle_s3_rec(rec):
    seen = read_s3_event(rec)

    if seen["bucket"] != INPUT_BUCKET or not seen["key"].startswith(INPUT_PREFIX):
        return

    pk = job_pk_for(seen)
    out_key = make_out_key(seen["key"])
    seen["output_key"] = out_key

    job = create_job_if_new(pk, seen)
    status = job.get("status")

    if status == "COMPLETED":
        return

    if not file_is_supported(seen["key"]):
        mark_done(pk, "SKIPPED", "unsupported_extension", out_key)
        return

    if output_exists(out_key):
        mark_done(pk, "COMPLETED", "", out_key)
        return

    claim_job(pk)

    try:
        source = head_source(seen)
        if source["size"] > MAX_BYTES:
            raise BadSource(f"input_too_large:{source['size']}")

        update_source_bits(pk, source)
        grab_rate_slot()
        clean_bytes, clean_type = call_backgrounderase(seen)
        put_clean_file(seen, out_key, clean_bytes, clean_type, pk)
        mark_done(pk, "COMPLETED", "", out_key)
    except BadSource as err:
        mark_done(pk, "FAILED", str(err), out_key)
    except Exception as err:
        note_retry(pk, err)
        raise


def read_s3_event(rec):
    obj = rec["s3"]["object"]
    return {
        "bucket": rec["s3"]["bucket"]["name"],
        "key": urllib.parse.unquote_plus(obj["key"]),
        "version_id": obj.get("versionId") or "null",
        "etag": (obj.get("eTag") or "").strip('"'),
        "size": int(obj.get("size") or 0),
        "event_name": rec.get("eventName", ""),
    }


def job_pk_for(seen):
    return f"JOB#{seen['bucket']}#{seen['key']}#{seen['version_id']}"


def make_out_key(key):
    rel = key[len(INPUT_PREFIX) :] if key.startswith(INPUT_PREFIX) else key
    path = PurePosixPath(rel)
    name = path.name
    stem = name.rsplit(".", 1)[0] if "." in name else name
    folder = "" if str(path.parent) == "." else f"{path.parent}/"
    return f"{OUTPUT_PREFIX}{folder}{stem}.png"


def file_is_supported(key):
    return PurePosixPath(key).suffix.lower() in SUPPORTED_EXTS


def s3_params(seen):
    params = {"Bucket": seen["bucket"], "Key": seen["key"]}
    if seen["version_id"] != "null":
        params["VersionId"] = seen["version_id"]
    return params


def now_s():
    return int(time.time())


def create_job_if_new(pk, seen):
    stamp = now_s()
    item = {
        "PK": pk,
        "status": "PENDING",
        "input_bucket": seen["bucket"],
        "input_key": seen["key"],
        "input_version_id": seen["version_id"],
        "input_etag": seen["etag"],
        "input_size": seen["size"],
        "output_bucket": OUTPUT_BUCKET,
        "output_key": seen["output_key"],
        "attempt_count": 0,
        "last_error": "",
        "created_at": stamp,
        "updated_at": stamp,
        "lease_expires_at": 0,
        "ttl": stamp + JOB_TTL_SECONDS,
    }

    try:
        jobs.put_item(Item=item, ConditionExpression="attribute_not_exists(PK)")
        return item
    except ClientError as err:
        if err.response["Error"].get("Code") != "ConditionalCheckFailedException":
            raise

    got = jobs.get_item(Key={"PK": pk}, ConsistentRead=True)
    return got.get("Item") or item


def claim_job(pk):
    stamp = now_s()
    try:
        jobs.update_item(
            Key={"PK": pk},
            UpdateExpression=(
                "SET #s = :progress, updated_at = :now, "
                "lease_expires_at = :lease ADD attempt_count :one"
            ),
            ConditionExpression=(
                "#s = :pending OR #s = :failed OR "
                "(#s = :progress AND lease_expires_at < :now)"
            ),
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={
                ":pending": "PENDING",
                ":failed": "FAILED",
                ":progress": "IN_PROGRESS",
                ":now": stamp,
                ":lease": stamp + LEASE_SECONDS,
                ":one": 1,
            },
        )
    except ClientError as err:
        if err.response["Error"].get("Code") == "ConditionalCheckFailedException":
            raise TryAgain("job_locked")
        raise


def head_source(seen):
    try:
        got = s3.head_object(**s3_params(seen))
        return {
            "size": int(got["ContentLength"]),
            "etag": got.get("ETag", "").strip('"'),
            "content_type": got.get("ContentType", ""),
        }
    except ClientError as err:
        code = err.response["Error"].get("Code")
        if code in {"404", "NoSuchKey", "NotFound"}:
            raise BadSource("source_missing")
        raise TryAgain(f"s3_head_failed:{code}")


def update_source_bits(pk, source):
    jobs.update_item(
        Key={"PK": pk},
        UpdateExpression="SET input_size = :size, input_etag = :etag, updated_at = :now",
        ExpressionAttributeValues={
            ":size": source["size"],
            ":etag": source["etag"],
            ":now": now_s(),
        },
    )


def output_exists(out_key):
    try:
        s3.head_object(Bucket=OUTPUT_BUCKET, Key=out_key)
        return True
    except ClientError as err:
        code = err.response["Error"].get("Code")
        if code in {"404", "NoSuchKey", "NotFound"}:
            return False
        raise TryAgain(f"s3_output_head_failed:{code}")


def grab_rate_slot():
    while True:
        now_ms = int(time.time() * 1000)
        got = ddb_raw.get_item(
            TableName=RATE_LIMIT_TABLE,
            Key={"PK": {"S": RATE_PK}},
            ConsistentRead=True,
        )
        item = got.get("Item", {})
        old_next = int(item.get("next_allowed_at_ms", {}).get("N", now_ms))
        slot_start = max(now_ms, old_next)
        new_next = slot_start + 1000

        if now_ms < old_next:
            time.sleep(min((old_next - now_ms) / 1000, 5))

        try:
            ddb_raw.update_item(
                TableName=RATE_LIMIT_TABLE,
                Key={"PK": {"S": RATE_PK}},
                UpdateExpression="SET next_allowed_at_ms = :new_next",
                ConditionExpression=(
                    "attribute_not_exists(next_allowed_at_ms) "
                    "OR next_allowed_at_ms = :old_next"
                ),
                ExpressionAttributeValues={
                    ":old_next": {"N": str(old_next)},
                    ":new_next": {"N": str(new_next)},
                },
            )
            return
        except ClientError as err:
            if err.response["Error"].get("Code") != "ConditionalCheckFailedException":
                raise
            time.sleep(random.uniform(0.05, 0.3))


def call_backgrounderase(seen):
    url = s3.generate_presigned_url(
        "get_object",
        Params=s3_params(seen),
        ExpiresIn=PRESIGN_SECONDS,
    )
    fields = {
        "image_url": url,
        "format": "png",
        "channels": "rgba",
    }
    body, content_type = make_multipart(fields)
    req = urllib.request.Request(
        API_URL,
        data=body,
        method="POST",
        headers={
            "x-api-key": be_api_key(),
            "Content-Type": content_type,
            "Accept": "image/*",
        },
    )

    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            img = resp.read()
            got_type = resp.headers.get("Content-Type", "")
    except urllib.error.HTTPError as err:
        api_body = err.read().decode("utf-8", "ignore")[:800]
        if err.code == 429:
            wait = retry_after_seconds(err.headers.get("Retry-After"))
            if wait:
                time.sleep(min(wait, 15))
            raise TryAgain(f"backgrounderase_429:{api_body}")
        if err.code in {400, 413, 415, 422}:
            raise BadSource(f"backgrounderase_rejected:{err.code}:{api_body}")
        if err.code == 408 or 500 <= err.code <= 599:
            raise TryAgain(f"backgrounderase_retryable:{err.code}:{api_body}")
        raise TryAgain(f"backgrounderase_unexpected:{err.code}:{api_body}")
    except urllib.error.URLError as err:
        raise TryAgain(f"backgrounderase_network:{err.reason}")

    if not img or not got_type.startswith("image/"):
        raise TryAgain(f"backgrounderase_non_image:{got_type}")

    return img, got_type


def make_multipart(fields):
    boundary = f"----be-{uuid.uuid4().hex}"
    parts = []
    for name, value in fields.items():
        parts.append(f"--{boundary}".encode())
        parts.append(f'Content-Disposition: form-data; name="{name}"'.encode())
        parts.append(b"")
        parts.append(str(value).encode())
    parts.append(f"--{boundary}--".encode())
    parts.append(b"")
    return b"\r\n".join(parts), f"multipart/form-data; boundary={boundary}"


def retry_after_seconds(value):
    if not value:
        return 0
    try:
        return max(0, int(value))
    except ValueError:
        return 0


def be_api_key():
    global be_key_cache
    if be_key_cache:
        return be_key_cache

    raw = secrets.get_secret_value(SecretId=SECRET_NAME).get("SecretString", "")
    try:
        parsed = json.loads(raw)
        be_key_cache = parsed.get("BACKGROUNDERASE_API_KEY") or parsed.get("api_key") or raw
    except json.JSONDecodeError:
        be_key_cache = raw.strip()
    return be_key_cache


def put_clean_file(seen, out_key, img, content_type, pk):
    s3.put_object(
        Bucket=OUTPUT_BUCKET,
        Key=out_key,
        Body=img,
        ContentType="image/png" if content_type.startswith("image/") else content_type,
        Metadata={
            "source-bucket": seen["bucket"],
            "source-key": urllib.parse.quote(seen["key"], safe="/-_.~"),
            "source-version-id": seen["version_id"],
            "backgrounderase-job-id": pk,
        },
    )


def mark_done(pk, status, err_text, out_key):
    stamp = now_s()
    expression = (
        "SET #s = :status, updated_at = :now, last_error = :err, "
        "lease_expires_at = :zero, output_key = :out"
    )
    values = {
        ":status": status,
        ":now": stamp,
        ":err": err_text[:900],
        ":zero": 0,
        ":out": out_key,
    }

    if status == "COMPLETED":
        expression += ", completed_at = :now"

    jobs.update_item(
        Key={"PK": pk},
        UpdateExpression=expression,
        ExpressionAttributeNames={"#s": "status"},
        ExpressionAttributeValues=values,
    )


def note_retry(pk, err):
    jobs.update_item(
        Key={"PK": pk},
        UpdateExpression="SET updated_at = :now, last_error = :err, lease_expires_at = :zero",
        ExpressionAttributeValues={
            ":now": now_s(),
            ":err": str(err)[:900],
            ":zero": 0,
        },
    )

bash <<'BASH'
set -euo pipefail

INPUT_BUCKET=backgrounderase-pipeline-input
OUTPUT_BUCKET=backgrounderase-pipeline-output
MAIN_QUEUE_NAME=backgrounderase-image-jobs
ROLE_NAME=backgrounderase-image-worker-role
JOBS_TABLE=BackgroundEraseJobs
RATE_LIMIT_TABLE=BackgroundEraseRateLimit
SECRET_NAME=backgrounderase/api-key

REGION=$(aws configure get region)
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
if [ -z "$REGION" ]; then
  echo "Set AWS_REGION or configure a default AWS CLI region first." >&2
  exit 1
fi
MAIN_QUEUE_URL=$(aws sqs get-queue-url \
  --queue-name "$MAIN_QUEUE_NAME" \
  --query QueueUrl \
  --output text)
MAIN_QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' \
  --output text)

cat > lambda-trust-policy.json <<'JSON'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
JSON

if ! aws iam get-role --role-name "$ROLE_NAME" >/dev/null 2>&1; then
  aws iam create-role \
    --role-name "$ROLE_NAME" \
    --assume-role-policy-document file://lambda-trust-policy.json
fi

aws iam wait role-exists --role-name "$ROLE_NAME"

cat > lambda-worker-policy.json <<JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:GetObjectVersion"],
      "Resource": "arn:aws:s3:::$INPUT_BUCKET/input/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:HeadObject"],
      "Resource": "arn:aws:s3:::$OUTPUT_BUCKET/output/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:ChangeMessageVisibility",
        "sqs:GetQueueAttributes"
      ],
      "Resource": "$MAIN_QUEUE_ARN"
    },
    {
      "Effect": "Allow",
      "Action": ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem"],
      "Resource": [
        "arn:aws:dynamodb:$REGION:$ACCOUNT_ID:table/$JOBS_TABLE",
        "arn:aws:dynamodb:$REGION:$ACCOUNT_ID:table/$RATE_LIMIT_TABLE"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "arn:aws:secretsmanager:$REGION:$ACCOUNT_ID:secret:$SECRET_NAME-*"
    },
    {
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": "*"
    }
  ]
}
JSON

aws iam put-role-policy \
  --role-name "$ROLE_NAME" \
  --policy-name backgrounderase-image-worker-inline \
  --policy-document file://lambda-worker-policy.json

sleep 10

aws iam get-role --role-name "$ROLE_NAME" --query 'Role.Arn' --output text
BASH

After the role exists and lambda_worker.py is saved locally, deploy the function. Do not connect the SQS trigger yet; give the function its environment first so the first queued message does not become a preventable error. The deploy block waits for the function to become active before setting reserved concurrency, because Lambda also has a short "resource is still updating" window.

bash <<'BASH'
set -euo pipefail

ROLE_NAME=backgrounderase-image-worker-role
ROLE_ARN=$(aws iam get-role \
  --role-name "$ROLE_NAME" \
  --query 'Role.Arn' \
  --output text)

zip function.zip lambda_worker.py

if aws lambda get-function --function-name backgrounderase-image-worker >/dev/null 2>&1; then
  aws lambda update-function-code \
    --function-name backgrounderase-image-worker \
    --zip-file fileb://function.zip

  aws lambda wait function-updated \
    --function-name backgrounderase-image-worker
else
  aws lambda create-function \
    --function-name backgrounderase-image-worker \
    --runtime python3.12 \
    --handler lambda_worker.lambda_handler \
    --role "$ROLE_ARN" \
    --timeout 300 \
    --memory-size 1024 \
    --zip-file fileb://function.zip

  aws lambda wait function-active \
    --function-name backgrounderase-image-worker
fi

aws lambda put-function-concurrency \
  --function-name backgrounderase-image-worker \
  --reserved-concurrent-executions 3
BASH

Then add the environment variables shown above in the Lambda Console. If you are staying in the CLI, create the API key secret and set the function environment like this. Keep the API key in Secrets Manager under backgrounderase/api-key; do not paste it directly into the Lambda environment.

bash <<'BASH'
set -euo pipefail

read -rsp "BackgroundErase API key: " BE_API_KEY < /dev/tty
printf '\n'

if aws secretsmanager describe-secret --secret-id backgrounderase/api-key >/dev/null 2>&1; then
  aws secretsmanager put-secret-value \
    --secret-id backgrounderase/api-key \
    --secret-string "$BE_API_KEY"
else
  aws secretsmanager create-secret \
  --name backgrounderase/api-key \
    --secret-string "$BE_API_KEY"
fi

aws lambda update-function-configuration \
  --function-name backgrounderase-image-worker \
  --environment 'Variables={INPUT_BUCKET=backgrounderase-pipeline-input,INPUT_PREFIX=input/,OUTPUT_BUCKET=backgrounderase-pipeline-output,OUTPUT_PREFIX=output/,JOBS_TABLE=BackgroundEraseJobs,RATE_LIMIT_TABLE=BackgroundEraseRateLimit,BACKGROUNDERASE_SECRET_NAME=backgrounderase/api-key,BACKGROUNDERASE_API_URL=https://api.backgrounderase.com/v2,PRESIGNED_URL_EXPIRATION_SECONDS=900,MAX_INPUT_FILE_SIZE_BYTES=30000000}'

aws lambda wait function-updated \
  --function-name backgrounderase-image-worker
BASH

Once the configuration update succeeds, connect the main SQS queue as the trigger. Use batch size 1 and maximum concurrency 3 on the event source mapping. If you rerun this block later, it updates the existing mapping instead of trying to create a duplicate one.

bash <<'BASH'
set -euo pipefail

MAIN_QUEUE_NAME=backgrounderase-image-jobs
MAIN_QUEUE_URL=$(aws sqs get-queue-url \
  --queue-name "$MAIN_QUEUE_NAME" \
  --query QueueUrl \
  --output text)
MAIN_QUEUE_ARN=$(aws sqs get-queue-attributes \
  --queue-url "$MAIN_QUEUE_URL" \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' \
  --output text)

MAPPING_UUID=$(aws lambda list-event-source-mappings \
  --function-name backgrounderase-image-worker \
  --event-source-arn "$MAIN_QUEUE_ARN" \
  --query 'EventSourceMappings[0].UUID' \
  --output text)

if [ "$MAPPING_UUID" = "None" ] || [ -z "$MAPPING_UUID" ]; then
  aws lambda create-event-source-mapping \
    --function-name backgrounderase-image-worker \
    --event-source-arn "$MAIN_QUEUE_ARN" \
    --batch-size 1 \
    --scaling-config MaximumConcurrency=3
else
  aws lambda update-event-source-mapping \
    --uuid "$MAPPING_UUID" \
    --batch-size 1 \
    --scaling-config MaximumConcurrency=3
fi
BASH

08

Lambda worker explanation

The worker file is doing several jobs that are easy to miss if you only look at the AWS wiring: global rate limiting, presigned URL generation, API response validation, idempotent output writes, and failure classification. You do not need to run more setup commands in this section. This is the map for debugging and modifying lambda_worker.py.

Rate limiting

The concurrency cap handles maximum simultaneous connections. It does not enforce one request per second. Three Lambdas can still wake up at the same time and send three requests in the same second, which is how you get avoidable 429s and then a little retry bonfire.

Use a DynamoDB-backed next-slot limiter. In the worker file above, that is the grab_rate_slot() function. Every worker has to reserve the next global API slot before it calls BackgroundErase. If another worker wins the conditional update, the loser waits and tries again.

To verify it is working, watch the Lambda logs during a burst. You should see workers wait around the rate slot instead of all calling the API at once. If a deployment bug writes a timestamp far into the future, reset the limiter row by setting next_allowed_at_ms back to 0 in BackgroundEraseRateLimit.

This limiter is stricter than a true token bucket because it does not use the full burst capacity of 10. That is acceptable for a production tutorial because correctness matters more than squeezing out the first ten requests. If you need the burst behavior, store token count and last refill time in DynamoDB and use conditional updates around both values.

The API can still return 429. Respect Retry-After when it is present, use exponential backoff with jitter, and let SQS retry later when the response is clearly transient.

Presigned URLs and API calls

Do not put presigned URLs in SQS or DynamoDB. They can expire while waiting in the queue, especially during a large import. Store only the bucket, key, version ID, ETag, size, and job metadata. Generate the URL immediately before the API call.

This happens inside call_backgrounderase(seen) in lambda_worker.py. The function generates the presigned URL, builds the multipart request, sends image_url, format=png, and channels=rgba, then rejects empty or non-image responses.

def call_backgrounderase(seen):
    url = s3.generate_presigned_url(
        "get_object",
        Params=s3_params(seen),
        ExpiresIn=PRESIGN_SECONDS,
    )
    fields = {
        "image_url": url,
        "format": "png",
        "channels": "rgba",
    }
    body, content_type = make_multipart(fields)
    req = urllib.request.Request(
        API_URL,
        data=body,
        method="POST",
        headers={
            "x-api-key": be_api_key(),
            "Content-Type": content_type,
            "Accept": "image/*",
        },
    )
    with urllib.request.urlopen(req, timeout=120) as resp:
        img = resp.read()
        got_type = resp.headers.get("Content-Type", "")

    if not img or not got_type.startswith("image/"):
        raise TryAgain(f"backgrounderase_non_image:{got_type}")

    return img, got_type

For the basic background-removal output, send image_url, format=png, and channels=rgba. If you need a white-background JPEG, thumbnail WebP, or a different crop policy, that belongs in your output recipe. This article is focused on the pipeline, not every possible image setting.

Output writes

Write the processed file to the output bucket or prefix with a predictable key. A simple first pass is to mirror the input key and change the extension to .png. In a product catalog, you may also include customer ID, SKU, output recipe, dimensions, or a source version hash.

In the worker file, make_out_key() decides the output path and put_clean_file() uploads the result. The key detail is that output_exists() runs before the API call. That makes retries cheaper when the API succeeded but the final job update failed.

def put_clean_file(seen, out_key, img, content_type, pk):
    s3.put_object(
        Bucket=OUTPUT_BUCKET,
        Key=out_key,
        Body=img,
        ContentType="image/png" if content_type.startswith("image/") else content_type,
        Metadata={
            "source-bucket": seen["bucket"],
            "source-key": urllib.parse.quote(seen["key"], safe="/-_.~"),
            "source-version-id": seen["version_id"],
            "backgrounderase-job-id": pk,
        },
    )

Before calling the API, check whether the expected output object already exists. This matters for idempotency and for the annoying edge case where the API succeeds but the final S3 upload or DynamoDB update fails. On retry, you want to notice the output and mark the job complete instead of paying for another identical API call.

Failures and retries

Failure handling should be boring and explicit. Split errors into transient, permanent, and ambiguous. Then make the Lambda either raise so SQS retries the message, or mark the job terminal and return success so the message is removed.

In lambda_worker.py, permanent failures raise BadSource and are marked FAILED or SKIPPED. Retryable failures raise a normal exception after note_retry() records the last error, which causes Lambda to return a failure to SQS. SQS then redelivers the message until Maximum receives is hit.

Transient:
  429 rate limit
  408 timeout
  500-599 API response
  network timeout
  temporary S3 or DynamoDB failure
  -> raise so SQS retries later

Permanent:
  unsupported file type
  object too large
  invalid image
  known API validation error
  -> mark FAILED or SKIPPED and return success

Ambiguous:
  API returned non-image body
  output upload failed after API success
  unexpected exception
  DynamoDB state conflict
  -> retry a few times, then inspect in the DLQ

Permanent errors should not keep retrying for fourteen days just because the queue is technically capable of it. Mark corrupt files, unsupported formats, objects above your maximum size, and known API validation errors as FAILED or SKIPPED in DynamoDB, then return success.

Ambiguous errors deserve a few retries and then the DLQ. When a message lands in the DLQ, inspect the DynamoDB job record and the last error before replaying it. Blind DLQ replay is how teams rediscover the same permanent error with more confidence and less patience.

09

Monitoring and alarms

A production pipeline should tell you two things quickly: is work draining, and are failures accumulating somewhere quiet? CloudWatch metrics are not glamorous, but the queue age graph will save you from pretending a million-image import is "almost done" when it is actually on day nine.

SQS ApproximateNumberOfMessagesVisible > expected backlog
SQS ApproximateAgeOfOldestMessage > 10 days
DLQ ApproximateNumberOfMessagesVisible > 0
Lambda Errors > 0
Lambda Throttles > 0
Lambda Duration p95 near timeout
Lambda ConcurrentExecutions > 3
DynamoDB ReadThrottleEvents or WriteThrottleEvents > 0
BackgroundErase API 200 count drops unexpectedly
BackgroundErase API 429 count > 0
BackgroundErase API 5xx count > 0
Jobs COMPLETED count falls below expected drain rate
Jobs FAILED count rises above normal baseline

The most important alarm is ApproximateAgeOfOldestMessage > 10 days. SQS message retention maxes out at 14 days. If the oldest message is already 10 days old, you are running out of runway and need to raise quota, add a reconciliation job, or stop feeding the queue until it catches up.

Also alarm when the DLQ has any visible messages. A DLQ is not a trash can. It is a shelf where bad jobs wait for a human to decide whether they are replayable, permanently failed, or evidence that your code did something exciting.

Create at least the queue-age alarm and the DLQ alarm before you let real uploads hit the bucket. This CLI block creates an SNS topic, asks for an alert email address, subscribes that email, and uses the returned topic ARN as the alarm action. AWS will send a confirmation email; the alarms will not notify you until that subscription is confirmed.

bash <<'BASH'
set -euo pipefail

MAIN_QUEUE_NAME=backgrounderase-image-jobs
DLQ_NAME=backgrounderase-image-jobs-dlq

read -rp "Alert email address: " ALERT_EMAIL < /dev/tty

SNS_TOPIC_ARN=$(aws sns create-topic \
  --name backgrounderase-pipeline-alerts \
  --query TopicArn \
  --output text)

aws sns subscribe \
  --topic-arn "$SNS_TOPIC_ARN" \
  --protocol email \
  --notification-endpoint "$ALERT_EMAIL"

aws cloudwatch put-metric-alarm \
  --alarm-name backgrounderase-main-queue-oldest-message-over-10-days \
  --namespace AWS/SQS \
  --metric-name ApproximateAgeOfOldestMessage \
  --dimensions Name=QueueName,Value="$MAIN_QUEUE_NAME" \
  --statistic Maximum \
  --period 300 \
  --evaluation-periods 1 \
  --threshold 864000 \
  --comparison-operator GreaterThanThreshold \
  --treat-missing-data notBreaching \
  --alarm-actions "$SNS_TOPIC_ARN"

aws cloudwatch put-metric-alarm \
  --alarm-name backgrounderase-dlq-has-messages \
  --namespace AWS/SQS \
  --metric-name ApproximateNumberOfMessagesVisible \
  --dimensions Name=QueueName,Value="$DLQ_NAME" \
  --statistic Maximum \
  --period 300 \
  --evaluation-periods 1 \
  --threshold 0 \
  --comparison-operator GreaterThanThreshold \
  --treat-missing-data notBreaching \
  --alarm-actions "$SNS_TOPIC_ARN"
BASH

10

Limitations of this setup

SQS can absorb a large upload burst. That does not mean the images process instantly. With the default BackgroundErase quota in this article, the pipeline should intentionally drain at about one image per second per account unless your account has a higher API quota.

This table is the part people like to skip because it is rude. A million-image backlog is about 11.6 days at one request per second. That is technically inside 14-day SQS retention if nothing else goes wrong, but it is close enough to the edge that I would want a higher API quota or a backfill strategy before calling it safe.

If you expect backlogs larger than what can drain within retention, pick one or more of these: request a higher BackgroundErase API quota, use S3 Inventory plus a scheduled backfill or reconciliation job, keep DynamoDB as the durable job ledger and periodically requeue pending work, or run a separate batch processing job for historical imports.

11

When to use a different architecture

S3 to SQS to Lambda is a good default for continuous uploads, short jobs, and an external API bottleneck. It is not the only good answer. Pick the architecture based on the type of work, not because an architecture diagram looked tidy.