Document AI Pipeline Tutorial: S3 PDF to DynamoDB

[!NOTE] TL;DR: Drop a PDF in S3, read a validated record in DynamoDB a minute later. This tutorial builds a complete document AI pipeline on AWS (Textract OCR, Bedrock extraction, Pydantic validation, Step Functions waiting) with runnable code and prices verified in November 2026.

In the beginning was the PDF, and it was already unreadable. A client of mine parks roughly 40,000 scanned invoices a year in an S3 bucket, where exactly zero machines read them before a human renames the files by hand. (That human has been promoted since, so the pile grows.)

The question of this tutorial is narrow: how do you turn that bucket into a document AI pipeline that a Python engineer can reproduce on a fresh AWS account in about an hour, with code that runs as-is? Every quota, price and model ID below was checked against the AWS documentation and the published price list in November 2026, because these numbers rot fast. If you want the conceptual map first, I wrote one in an earlier article, and there is a deeper industrial IDP angle here as well.

The mission, and the end state

You build five things, in order: an S3 inbox, a DynamoDB table, two IAM roles, one Lambda package with two handlers, and one Step Functions state machine. The end state is simple. A PDF lands in S3, the state machine starts an asynchronous Textract job, waits, collects the text blocks, asks Bedrock to turn them into typed invoice fields, validates those fields with Pydantic, and writes the item to DynamoDB. Anything that fails validation is not lost: it lands in the same table flagged NEEDS_REVIEW, ready for a human. If you'd rather have someone build and operate this with you, my services page describes how I work.

Prerequisites: AWS CLI v2, Python 3.12, and an account where Bedrock model access to Amazon Nova Lite is enabled in us-east-1. Textract gives 100 free pages per month on the Forms and Tables features, enough for a first test.

Think of the pipeline as a mail sorting room. S3 is the loading dock, Textract the clerk who reads the envelopes, Bedrock the clerk who fills in the form, Step Functions the foreman who does the waiting and the retrying, and DynamoDB the filing cabinet at the end of the room.

flowchart LR
    A["PDF uploaded to S3 inbox"] -->|"s3:ObjectCreated"| B["starter_lambda"]
    B -->|"StartExecution"| C["Step Functions state machine"]
    C --> D["Textract StartDocumentAnalysis"]
    D --> E["Wait 20 s"]
    E --> F["GetDocumentAnalysis"]
    F -->|"IN_PROGRESS"| E
    F -->|"SUCCEEDED"| G["extractor_lambda"]
    G --> H["Bedrock Converse + Pydantic validation"]
    H --> I["DynamoDB putItem, status PROCESSED"]
    H -->|"validation error"| J["DynamoDB putItem, status NEEDS_REVIEW"]

Step 1: The inbox bucket and the extraction table

The goal here is two named resources, nothing more.

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
aws s3api create-bucket --bucket docai-inbox-$ACCOUNT_ID --region us-east-1
aws dynamodb create-table --table-name docai_extractions \
  --attribute-definitions AttributeName=document_id,AttributeType=S \
  --key-schema AttributeName=document_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST

Expected result: a bucket in us-east-1 and a table with a single hash key on document_id, on-demand billing. The region choice is not cosmetic. Textract only reads buckets located in the region of the endpoint you call, so do not mix regions later.

Step 2: Two roles, no more

The goal is least-privilege IAM for the two execution planes. One role serves both Lambda handlers, the other serves the state machine, which calls Textract, your extractor and DynamoDB directly through its aws-sdk integrations.

cat > lambda-trust.json <<'EOF'
{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole"}]}
EOF
cat > lambda-policy.json <<'EOF'
{"Version": "2012-10-17", "Statement": [
  {"Effect": "Allow", "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"], "Resource": "arn:aws:logs:*:*:*"},
  {"Effect": "Allow", "Action": "states:StartExecution", "Resource": "arn:aws:states:*:*:stateMachine:docai-pipeline"},
  {"Effect": "Allow", "Action": ["textract:GetDocumentAnalysis", "bedrock:InvokeModel"], "Resource": "*"}
]}
EOF
aws iam create-role --role-name docai-lambda --assume-role-policy-document file://lambda-trust.json
aws iam put-role-policy --role-name docai-lambda --policy-name inline --policy-document file://lambda-policy.json
cat > states-trust.json <<'EOF'
{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": {"Service": "states.amazonaws.com"}, "Action": "sts:AssumeRole"}]}
EOF
cat > states-policy.json <<'EOF'
{"Version": "2012-10-17", "Statement": [
  {"Effect": "Allow", "Action": ["textract:StartDocumentAnalysis", "textract:GetDocumentAnalysis"], "Resource": "*"},
  {"Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::docai-inbox-*/*"},
  {"Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": "arn:aws:lambda:*:*:function:docai-extractor"},
  {"Effect": "Allow", "Action": "dynamodb:PutItem", "Resource": "arn:aws:dynamodb:*:*:table/docai_extractions"}
]}
EOF
aws iam create-role --role-name docai-states --assume-role-policy-document file://states-trust.json
aws iam put-role-policy --role-name docai-states --policy-name inline --policy-document file://states-policy.json
sleep 10

Expected result: two role ARNs, and a short pause for IAM propagation. The states role carries s3:GetObject on the inbox because Textract reads your PDF with the permissions of the caller.

Step 3: The extractor, where Pydantic arbitrates

The goal is one Lambda that collects the Textract job, renders it as text, and asks Bedrock for typed fields. I chose asynchronous Textract deliberately: the synchronous API caps PDFs at one page and 10 MB, while the asynchronous API accepts up to 500 MB and 3000 pages, and keeps results for 7 days before they expire.

On the Bedrock side I use the Converse API with outputConfig, the structured-output feature described in the AWS structured outputs post. Support depends on the model; all Claude models are covered since October 2026, and if your target model rejects outputConfig, the classic fallback is a forced tool call read from the toolUse block. Whatever the model returns, Pydantic re-validates it. The schema gives the response its shape; my validators give it its meaning.

import json
from datetime import datetime
import boto3
from pydantic import BaseModel, Field, field_validator
textract = boto3.client("textract")
bedrock = boto3.client("bedrock-runtime")
MODEL_ID = "us.amazon.nova-lite-v1:0"
SCHEMA = json.dumps({
    "type": "object",
    "properties": {
        "vendor_name": {"type": "string", "description": "Seller or issuing company"},
        "invoice_number": {"type": "string", "description": "Invoice reference"},
        "invoice_date": {"type": "string", "description": "Issue date, YYYY-MM-DD"},
        "total_amount": {"type": "number", "description": "Grand total, taxes included"},
        "currency": {"type": "string", "description": "ISO 4217 code, e.g. EUR"}},
    "required": ["vendor_name", "invoice_number", "invoice_date", "total_amount", "currency"],
    "additionalProperties": False})
class InvoiceFields(BaseModel):
    vendor_name: str = Field(..., min_length=1, description="Seller or issuing company.")
    invoice_number: str = Field(..., min_length=1, description="Invoice reference.")
    invoice_date: str = Field(..., description="Issue date, YYYY-MM-DD.")
    total_amount: float = Field(..., gt=0, description="Grand total, taxes included.")
    currency: str = Field(..., description="ISO 4217 currency code.")
    @field_validator("invoice_date")
    @classmethod
    def check_date(cls, value: str) -> str:
        datetime.strptime(value, "%Y-%m-%d")
        return value
    @field_validator("currency")
    @classmethod
    def check_currency(cls, value: str) -> str:
        if len(value) != 3 or not value.isalpha():
            raise ValueError("currency must be a three-letter ISO 4217 code")
        return value.upper()
def collect_blocks(job_id: str) -> list[dict]:
    blocks: list[dict] = []
    token: str | None = None
    while True:
        kwargs: dict = {"JobId": job_id, "MaxResults": 1000}
        if token:
            kwargs["NextToken"] = token
        page = textract.get_document_analysis(**kwargs)
        blocks.extend(page.get("Blocks", []))
        token = page.get("NextToken")
        if not token:
            return blocks
def render_text(blocks: list[dict]) -> str:
    return "\n".join(b["Text"] for b in blocks if b["BlockType"] == "LINE")
def invoke_bedrock(text: str) -> InvoiceFields:
    response = bedrock.converse(
        modelId=MODEL_ID,
        system=[{"text": "You extract invoice fields from OCR text. Return only values present in the text. Never invent anything."}],
        messages=[{"role": "user", "content": [{"text": "Extract the invoice fields from this OCR text:\n" + text}]}],
        inferenceConfig={"maxTokens": 1024, "temperature": 0.0},
        outputConfig={"textFormat": {"type": "json_schema", "structure": {"jsonSchema": {"schema": SCHEMA, "name": "invoice_fields"}}}})
    raw = response["output"]["message"]["content"][0]["text"]
    return InvoiceFields.model_validate_json(raw)
def handler(event: dict, context: object) -> dict:
    job_id: str = event["job"]["JobId"]
    blocks = collect_blocks(job_id)
    fields = invoke_bedrock(render_text(blocks))
    return {"fields": fields.model_dump(mode="json"), "block_count": len(blocks), "needs_review": False}

Expected result: a function that either returns a fully validated InvoiceFields payload or raises, which is exactly what we want the state machine to see.

Step 4: The state machine that does the waiting

The goal is orchestration without notification plumbing. Textract normally announces job completion through SNS, but wiring an SNS topic and an SQS queue just to learn that a job finished is a lot of furniture for a first build. The state machine polls instead: start, wait 20 seconds, check, loop. I once spent a Friday evening watching a Lambda poll Textract in a while loop until the 15-minute guillotine fell on it; the state machine does that waiting for free now, and shows every attempt in its execution graph.

{
  "Comment": "PDF to structured data: async Textract, Bedrock extraction, DynamoDB",
  "StartAt": "StartAnalysis",
  "States": {
    "StartAnalysis": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:textract:startDocumentAnalysis",
      "Parameters": {
        "DocumentLocation": {"S3Object": {"Bucket.$": "$.bucket", "Name.$": "$.key"}},
        "FeatureTypes": ["FORMS", "TABLES"]
      },
      "ResultPath": "$.job",
      "Next": "WaitForTextract"
    },
    "WaitForTextract": {"Type": "Wait", "Seconds": 20, "Next": "CheckStatus"},
    "CheckStatus": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:textract:getDocumentAnalysis",
      "Parameters": {"JobId.$": "$.job.JobId"},
      "ResultPath": "$.status",
      "Next": "StatusChoice"
    },
    "StatusChoice": {
      "Type": "Choice",
      "Choices": [
        {"Variable": "$.status.JobStatus", "StringEquals": "IN_PROGRESS", "Next": "WaitForTextract"},
        {"Variable": "$.status.JobStatus", "StringEquals": "SUCCEEDED", "Next": "ExtractFields"}
      ],
      "Default": "JobFailed"
    },
    "ExtractFields": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {"FunctionName": "REPLACE_EXTRACTOR_ARN", "Payload.$": "$"},
      "ResultPath": "$.fields",
      "Retry": [{"ErrorEquals": ["Lambda.TooManyRequestsException", "Lambda.ServiceException"], "IntervalSeconds": 3, "BackoffRate": 2.0, "MaxAttempts": 2}],
      "Catch": [{"ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "MarkForReview"}],
      "Next": "SaveItem"
    },
    "SaveItem": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
      "Parameters": {
        "TableName": "docai_extractions",
        "Item": {
          "document_id": {"S.$": "$.key"},
          "status": {"S": "PROCESSED"},
          "job_id": {"S.$": "$.job.JobId"},
          "fields": {"M.$": "$.fields.Payload.fields"},
          "needs_review": {"BOOL.$": "$.fields.Payload.needs_review"},
          "processed_at": {"S.$": "$$.State.EnteredTime"}
        }
      },
      "End": true
    },
    "MarkForReview": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:dynamodb:putItem",
      "Parameters": {
        "TableName": "docai_extractions",
        "Item": {
          "document_id": {"S.$": "$.key"},
          "status": {"S": "NEEDS_REVIEW"},
          "error": {"S.$": "$.error.Error"},
          "processed_at": {"S.$": "$$.State.EnteredTime"}
        }
      },
      "End": true
    },
    "JobFailed": {"Type": "Fail", "Error": "TextractJobFailed"}
  }
}

Save this as statemachine.json. Although the polling loop burns a few extra state transitions, it removes two services from the architecture, and the 20-second wait dominates end-to-end latency anyway: a two or three page invoice is usually done on the first or second poll.

Step 5: Wiring the S3 event and the first run

The goal is a package, three deployments, one S3 trigger, and a test PDF. The starter handler is four lines of boto3: parse the S3 event, unquote the key, start the execution.

import json
import os
import time
import urllib.parse
import boto3
sfn = boto3.client("stepfunctions")
STATE_MACHINE_ARN = os.environ["STATE_MACHINE_ARN"]
def handler(event: dict, context: object) -> dict:
    s3_record = event["Records"][0]["s3"]
    bucket: str = s3_record["bucket"]["name"]
    key: str = urllib.parse.unquote_plus(s3_record["object"]["key"])
    execution_name: str = f"{key.replace('/', '-')[:60]}-{int(time.time())}"
    sfn.start_execution(stateMachineArn=STATE_MACHINE_ARN, name=execution_name, input=json.dumps({"bucket": bucket, "key": key}))
    return {"started": key}

mkdir -p build
pip install -q --platform manylinux2014_x86_64 --only-binary=:all: --target build pydantic
cp starter_lambda.py extractor_lambda.py build/
(cd build && zip -rq ../pipeline.zip .)
EXTRACTOR_ARN=$(aws lambda create-function --function-name docai-extractor \
  --runtime python3.12 --handler extractor_lambda.handler --zip-file fileb://pipeline.zip \
  --role arn:aws:iam::$ACCOUNT_ID:role/docai-lambda --timeout 120 --memory-size 512 \
  --query FunctionArn --output text)
sed "s|REPLACE_EXTRACTOR_ARN|$EXTRACTOR_ARN|" statemachine.json > statemachine.rendered.json
SF_ARN=$(aws stepfunctions create-state-machine --name docai-pipeline \
  --definition file://statemachine.rendered.json \
  --role-arn arn:aws:iam::$ACCOUNT_ID:role/docai-states \
  --query stateMachineArn --output text)
STARTER_ARN=$(aws lambda create-function --function-name docai-starter \
  --runtime python3.12 --handler starter_lambda.handler --zip-file fileb://pipeline.zip \
  --role arn:aws:iam::$ACCOUNT_ID:role/docai-lambda --timeout 30 --memory-size 256 \
  --environment "Variables={STATE_MACHINE_ARN=$SF_ARN}" --query FunctionArn --output text)
aws lambda add-permission --function-name docai-starter --statement-id s3-invoke \
  --action lambda:InvokeFunction --principal s3.amazonaws.com \
  --source-arn arn:aws:s3:::docai-inbox-$ACCOUNT_ID
cat > notification.json <<EOF
{"LambdaFunctionConfigurations": [{"LambdaFunctionArn": "$STARTER_ARN", "Events": ["s3:ObjectCreated:*"], "Filter": {"Key": {"Suffix": ".pdf"}}}]}
EOF
aws s3api put-bucket-notification-configuration --bucket docai-inbox-$ACCOUNT_ID --notification-configuration file://notification.json
aws s3 cp invoice.pdf s3://docai-inbox-$ACCOUNT_ID/invoices/invoice-001.pdf
aws dynamodb get-item --table-name docai_extractions --key '{"document_id": {"S": "invoices/invoice-001.pdf"}}'

Expected result: the get-item returns a PROCESSED row with five validated fields, typically after one or two polls. If it does not, the Step Functions execution graph names the guilty state, and each Lambda logs to its own CloudWatch group. Add one CloudWatch alarm on the state machine's ExecutionsFailed metric (greater than zero for five minutes) and you have the minimum honest observability.

The complete file layout

Everything you wrote fits in one small directory, which is the point: a reference implementation you can read in one sitting.

docai-pipeline/
├── starter_lambda.py      # S3 event -> StartExecution
├── extractor_lambda.py    # Textract blocks -> Bedrock -> Pydantic
├── statemachine.json      # the ASL definition, Step 4
├── notification.json      # S3 trigger, generated in Step 5
└── build/                 # zip payload with pydantic

What a thousand pages cost

Textract bills per page and per feature, at the same rate for synchronous and asynchronous calls. FORMS costs USD 0.05 per page and TABLES USD 0.015, so a FORMS plus TABLES analysis costs USD 0.065 per page, or 65 USD per 1000 pages at first-tier prices in us-east-1. Plain text detection would cost 1.50 USD per 1000 pages, but you lose the key-value structure that makes the extraction trivial.

{
  "type": "bar",
  "data": {
    "labels": ["DetectDocumentText", "Analyze LAYOUT", "Analyze TABLES", "Analyze QUERIES", "Analyze FORMS", "FORMS + TABLES"],
    "datasets": [{ "label": "USD per 1000 pages (us-east-1, first tier)", "data": [1.5, 4.0, 15.0, 15.0, 50.0, 65.0], "backgroundColor": ["#3b82f6"] }]
  },
  "options": {
    "responsive": true,
    "plugins": { "title": { "display": true, "text": "Textract list price per 1000 pages, us-east-1, November 2026" } },
    "scales": { "y": { "title": { "display": true, "text": "USD per 1000 pages" } } }
  }
}

On the Bedrock side, Nova Lite lists at USD 0.06 per million input tokens and USD 0.24 per million output. Assume roughly 3000 input tokens and 300 output per invoice, and 1000 invoices cost on the order of 0.25 USD. The model is noise in the bill; Textract is the line item. Figures come from the Textract pricing page, and the token estimate is my assumption, not a measured bill.

What could go wrong

The failure modes below are the ones I actually hit or verified in the Textract limits documentation.

That last row deserves a name: the zombie execution. An upload can trigger your starter twice, and a naive design would bill Textract twice for the same pages. Here idempotency comes free because the DynamoDB key is the S3 key.

[!WARNING] Structured output guarantees the shape of the JSON, never the truth of the values. A model can return a well-formed, schema-compliant, entirely invented total. Validate types with Pydantic, then validate plausibility in your own code.

Now the honest limits. I validated this pipeline on test invoices of a few pages, not under load. At thousands of documents a day, swap the polling loop for the SNS notification pattern and add a per-document idempotency key on the Textract call (ClientRequestToken), and be aware that a Bedrock model version change can degrade extraction quality silently, which no exception will ever tell you. The production pipelines I maintain on larger corpora use a Step Functions Map state at concurrency 20 per page, an SQS buffer at the entrance, and raw Textract JSON archived to S3 because DynamoDB items cap at 400 KB. If any of that sounds like your backlog, contact me and we'll scope it properly.

Final zoom

More generally, the durable idea in this pipeline is a separation of duties. Textract reads pixels, the model interprets text, the state machine does the waiting, and only a typed validator decides what deserves a column in your database. Keep those four jobs in four different hands, and the whole thing survives model swaps, price changes and your own Fridays. Go feed it a PDF!