Forem: Matia Rašetina

DocumentDB vs DynamoDB in a Serverless Stack: I Benchmarked Both So You Don't Have To

Matia Rašetina — Mon, 13 Apr 2026 07:00:00 +0000

If you Google or chat with any of the LLMs about “DynamoDB vs DocumentDB”, most answers are theoretical.

DynamoDB is supposed to be faster because it’s serverless. DocumentDB is supposed to be slower because it runs in a VPC, but no one is showing hard proof that this truly is the case. That’s why I built this experiment with 1000 requests with 200 concurrent users with two simple operations — writes and reads. And here are the results!

Full GitHub code is available here.

Experimental Setup

Both databases sit behind the same architecture: an API Gateway REST API invoking a Python 3.12 Lambda function inside a VPC, the Lambda connects to the database, executes the requested operation and returns the information back to our testing script.

For the DocumentDB stack, there are some additions — the DocumentDB instance needs to be inside a VPC, since it can only be set up in this way, and to use it, it requires credentials (username and password in this case). The database credentials are stored safely inside the Secrets Manager, which the Lambda will use to write and read from the DocumentDB instance.

DynamoDB is serverless by design — it scales automatically and you pay only for what you use. To make the comparison fair, I configured DocumentDB in Serverless mode as well (0.5–16 DCU, scaling on demand), so neither database has a fixed instance sitting idle between requests.

To make this experiment as reliable as possible, cold starts for Lambdas won’t be counted in the final result for both scenarios.

	DocumentDB	DynamoDB
Instance / Mode	Serverless (0.5–16 DCU)	Serverless (pay-per-request)
Region	us-east-1	us-east-1
Lambda runtime	Python 3.12	Python 3.12
Lambda memory	512 MB	512 MB
Lambda timeout	30 s	30 s
Lambda location	Private subnet, VPC	Private subnet, VPC
Client library	pymongo	boto3
Storage	Encrypted at rest	Managed by AWS
Test volume	1,000 requests	1,000 requests
Concurrency	200	200
Data	User records, ~400 bytes	Same schema, same payload

Another few details to optimize the Lambda runtime and keep the experiment fair:

MongoDB and DynamoDB clients are created outside the handler to allow connection reuse across invocations
DynamoDB client is configured to have ConsistentRead flag as False , to have quicker responses from the database

Here is the architecture diagram of this experiment, together with a short CDK snippet for creating a DocumentDB cluster and instance. The full code is available in the GitHub repository provided in the introduction.

# DocumentDB cluster
# Minimal 0.5 DCU configuration, maximum 16 DCU
self.docdb_cluster = docdb.CfnDBCluster(
    self,
    "DocDbCluster",
    master_username=self.docdb_secret.secret_value_from_json(
        "username"
    ).unsafe_unwrap(),
    master_user_password=self.docdb_secret.secret_value_from_json(
        "password"
    ).unsafe_unwrap(),
    db_subnet_group_name=subnet_group.ref,
    vpc_security_group_ids=[self.docdb_sg.security_group_id],
    storage_encrypted=True,
    engine_version="5.0.0",
    deletion_protection=False,
    serverless_v2_scaling_configuration=docdb.CfnDBCluster.ServerlessV2ScalingConfigurationProperty(
        min_capacity=0.5,
        max_capacity=16,
    ),
)
self.docdb_cluster.apply_removal_policy(RemovalPolicy.DESTROY)

# Serverless DocumentDB instance
docdb_instance = docdb.CfnDBInstance(
    self,
    "DocDbInstance",
    db_cluster_identifier=self.docdb_cluster.ref,
    db_instance_class="db.serverless",
    auto_minor_version_upgrade=True,
)
docdb_instance.add_dependency(self.docdb_cluster)

The Results

POST — Writing Data

This is the operation that creates a user record in the database, with a payload very similar to this one:

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "name": "user-47291",
  "timestamp": 1710249600,
  "payload": "xK9mN2pQr5sTvW7yZbCdFgHjL4nPqRsTuVwXyZ0123456789AbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWxYz01"
}

And here are the results:

Metric	DocumentDB	DynamoDB	Winner
P50	354.9 ms	358.4 ms	DocumentDB (-1%)
P90	523.9 ms	585.3 ms	DocumentDB (-10%)
P95	913.9 ms	602.3 ms	DynamoDB (-34%)
P99	977.4 ms	605.7 ms	DynamoDB (-38%)
Mean	347.4 ms	324.8 ms	DynamoDB (-7%)
Max	3,033 ms	905.4 ms	DynamoDB (-70%)
Throughput	217 RPS	508 RPS	DynamoDB (2.3x)
Failures	0 / 1000	0 / 1000	Tie

Looking at P50, both databases have very identical performance — DocumentDB at 354ms, DynamoDB at 358ms. Up to P90, DocumentDB holds a slight edge, but nothing really dramatic.

Now we get to P95, where we start to see the difference — DocumentDB jumps to 913ms, while DynamoDB handles traffic 34% better by staying around the 600ms mark. In addition, P99 further confirms the wide gap between the write performances of these two NoSQL databases, DocumentDB almost hits a second, while DynamoDB stays in the 600ms range. The max duration of the request is the most dramatic part, DocumentDB takes 3 seconds, while DynamoDB stays under a second for the same operation.

Regarding the throughput, DynamoDB beats DocumentDB by 2.3x, which is a huge difference since the rest of the configuration is identical — similar VPC, Lambda and payloads. This is the VPC + connection model showing up under write pressure.

Both databases hit 100% success rate with no dropped requests.

GET — Reading Data

However, this is where things get interesting — the scenario is simple, Lambda gets invoked and fetches the user record by ID inside the database.

For DocumentDB, that's a find_one by indexed field. For DynamoDB, a get_item by partition key.

Metric	DocumentDB	DynamoDB	Winner
P50	461.1 ms	466.4 ms	DocumentDB (-1%)
P90	796.1 ms	1,387.7 ms	DocumentDB (-43%)
P95	830.2 ms	1,427.4 ms	DocumentDB (-42%)
P99	2,102.2 ms	1,543.7 ms	DynamoDB (+36%)
Mean	443.8 ms	535.3 ms	DocumentDB (-17%)
Max	2,394.7 ms	1,786.3 ms	DynamoDB
Throughput	380 RPS	285 RPS	DocumentDB (1.3x)
Failures	0 / 1000	0 / 1000	Tie

This result truly surprised me. At P50, DynamoDB has a small edge on reads, which was expected, however the difference at P90 and later changed the way I look at DocumentDB, as it provided better read performance than DynamoDB by around 40% — if you look at P90, DocumentDB just stays below the 800ms mark, while DynamoDB almost hits the 1.4 second mark, and the situation is very similar for P95.

A possible explanation of why DocumentDB beat DynamoDB for reads is the buffer cache. DocumentDB keeps frequently accessed documents in memory on the instance. Once the cache is warm — which is exactly the steady-state condition in production — repeated reads on the same dataset are served from RAM rather than storage.

At P99, DynamoDB regains its composure and beats DocumentDB by being faster 36%. A likely explanation is connection pool exhaustion, as the slowest 1% of requests are possibly queued and are waiting for a pymongo connection to free up.

Regarding the throughput, DocumentDB processes the read requests faster by 1.3x.

Again, as before, both databases maintained 100% success rate.

What the Numbers Actually Tell You

For writes under load: DynamoDB is the clear winner here. The tail latency difference is too significant and is going to impact user experience in every way. If you need a database for ingesting events, logging and creating records which have unstructured data, DynamoDB is the definite winner.

For reads under load: DocumentDB wins, and by a meaningful margin at the tail. The throughput gap is also notable, as DocumentDB serves 33% more read requests per second in this test.

An important statistic to mention — P50. Why? Because both databases have it within 10% of each other in both operations, so if you stopped here, you might have an argument that they have identical performances. However, the true difference live in P95 and P99 stats, where your slowest users are impacted the most.

Pricing: What Does This Actually Cost?

Performance is only a part of your architectural decision. Here's what running each database actually costs at the scale tested.

Lambda (same for both)

Lambda pricing is identical regardless of which database you use:

Compute: $0.0000166667 per GB-second. At 512 MB and ~700ms average duration: roughly $0.006 per 1,000 requests
Requests: $0.20 per million invocations
At 1 million requests/month: ~$6 in compute + $0.20 in requests = ~$6.20/month

Lambda is not the cost driver here. The database is.

DynamoDB (On-Demand, us-east-1)

DynamoDB on-demand charges per request:

Writes: $0.625 per million write request units (1 WRU = 1 KB write)
Reads: $0.125 per million read request units (1 RRU = 4 KB strongly consistent read)
Storage: $0.25 per GB-month

For our ~400 byte payload (rounds up to 1 KB write, well under 4 KB read):

Volume	Writes	Reads	Storage (10GB)	Monthly Total
100K req/day (~3M/month)	$1.88	$0.38	$2.50	~$4.76
1M req/day (~30M/month)	$18.75	$3.75	$2.50	~$25
10M req/day (~300M/month)	$187.50	$37.50	$2.50	~$227

No instance cost, as DynamoDB is a serverless NoSQL database.

DocumentDB (Serverless, us-east-1)

Pricing dimensions:

Compute: ~$0.0822 per DCU-hour (Standard config, us-east-1)
Storage: $0.10 per GB-month
I/O: $0.20 per million requests
Minimum capacity: 0.5 DCU (~1 GB RAM equivalent)

Volume	Avg DCU	Compute	Storage (10GB)	I/O	Monthly Total
100K req/day (~3M/month)	0.5	$29.59	$1.00	$0.60	~$31.19
1M req/day (~30M/month)	1	$59.18	$1.00	$6.00	~$66.18
10M req/day (~300M/month)	2	$118.37	$1.00	$60.00	~$179.37

This table is based on my CloudWatch statistics, where the database with 200 concurrent request reached 9.15 DCU, and during testing my random testing to see if the database works, the average DCU was 4.66 as seen in the image below.

The Cost Crossover

At low volumes, DynamoDB is dramatically cheaper, however, at very high traffic volumes, DocumentDB's per-I/O pricing ($0.20/million) becomes competitive with DynamoDB's per-request pricing ($0.625/million writes).

For this benchmark workload, the crossover point appeared around 5–10M requests/day.

When to Use Each

Scenario	Recommendation
Low traffic, unpredictable spikes	DynamoDB — zero idle cost
High write concurrency	DynamoDB — better tail latency on writes
High read concurrency	DocumentDB — better tail latency and throughput on reads
Simple key-value access patterns	DynamoDB — purpose-built for this
Complex queries, aggregations	DocumentDB — native `$match`, `$group`, `$sort` pipeline
Existing MongoDB codebase	DocumentDB — minimal migration friction

Lessons Learned

Measure before assuming. From my experience, it was always assumed that DynamoDB always provides best performance in the serverless infrastructure, but as we’ve proven with the read results at high concurrency, DocumentDB’s warm connection pool and buffer cache provide better tail latency than DynamoDB on-demand.

Tail latency is the real benchmark. Both databases are within 10% of each other at P50. The decision lives at P95 and P99. For write-heavy apps, that tail strongly favours DynamoDB. For read-heavy apps, it favours DocumentDB.

VPC overhead is real but manageable. DocumentDB lives in a VPC. The TCP/TLS handshake on cold starts adds latency. These are architectural decisions you have to live with if you go with DocumentDB, knowing that your reads are going to be better for your tail users.

Reproducing This Yourself

The full CDK stack (Python) and benchmark script are available on GitHub. Deploy it in your own account and run the same test — the numbers will vary by region, instance size, and concurrency, which is exactly the point. Your workload will tell you which database wins for your use case.

# Deploy from the CDK folder
cd cdk && cdk deploy

# Seed the databases with random data
python scripts/seed_dynamodb.py --table-name benchmark-users --count 1000
python scripts/seed_documentdb.py --stack-name DocDbApiStack --region us-east-1 --count 1000

# Run the benchmark for write operations
python api_load_test.py \
--endpoint https://<API_ENDPOINT_DYNAMODB_STACK>/prod/users \
--requests 1000 --concurrency 200 \
--method POST --output results_dynamodb_post.json

python api_load_test.py \
--endpoint https://<API_ENDPOINT_DOCUMENTDB_STACK>/prod/users \
--requests 1000 --concurrency 200 \
--method POST --output results_docdb_post.json

# Run the benchmark for read operations
python api_load_test.py \
--endpoint https://<API_ENDPOINT_DYNAMODB_STACK>/prod/users \
--requests 1000 --concurrency 200 \
--method GET --output results_dynamodb_post.json

python api_load_test.py \
--endpoint https://<API_ENDPOINT_DOCUMENTDB_STACK>/prod/users \
--requests 1000 --concurrency 200 \
--method GET --output results_docdb_get.json

AWS Lambda PII Handling in Production: DynamoDB Field Encryption with KMS

Matia Rašetina — Mon, 06 Apr 2026 07:00:00 +0000

Handling Personally Identifiable Information (or PII for short) in AWS Lambda-backed systems is not difficult because AWS lacks security primitives. It is difficult because the default patterns in backend development encourage storing sensitive data in places where access control is only enforced at the infrastructure boundary — DynamoDB encryption at rest protects disks, but not application data, and once PII enters any of the database solutions inside the system, every principal with read access becomes a major threat.

This article demonstrates a production-tested pattern — encrypting sensitive data categorized as PII (user’s home address, email, name, surname, date of birth etc.) in AWS Lambda before they are stored in DynamoDB, using customer-managed KMS keys and infrastructure entirely defined in Python AWS CDK. The goal of this pattern is to ensure that if ever a data breach happens to your application, that the data is encrypted security inside DynamoDB, and that only chosen parts of the system can decrypt the data if necessary. This pattern can be adjusted to be used with any other database solution available on AWS.

This post will have most important snippets of the whole pattern, but the full code, where the Lambda handles all CRUD operations with DynamoDB, is available here.

Why are we going over this pattern?

There are many examples of existing serverless systems built on AWS Lambda and DynamoDB where some of the data, which gets stored inside the database, qualifies as PII and must not be stored in a human-readable format. As previously mentioned, the infrastructure in this pattern is defined using AWS CDK in Python, with a Lambda runtime in Python as well. Encryption keys are stored in AWS Key Management Service (KMS) and are customer-managed with explicit key policies, together with a strictly defined Lambda execution role.

This design does not attempt to solve encrypted search or complex querying over sensitive fields, as those problems require different tradeoffs. The scope here is to deliberately limit access to certain fields which are categorized as PII and could give you a legal headache in case of a data breach.

Moreover, DynamoDB server-side encryption is a must in any environment, but insufficient for PII or any field which needs protection by encryption. It protects against physical and low-level threats, but it does nothing to prevent plaintext access through IAM — any resource which has IAM access to the table and can scan or query the data, which can be a weakness in your infrastructure.

The only way to tip the scales in your favor here is to make sure that the sensitive fields are never stored in plaintext anywhere in your infrastructure. Once encryption happens in the application layer, any database solution becomes a persistence layer for ciphertext, and access to the data is no longer enough to expose PII. Data decryption process then becomes an explicit operation, which only authorized compute resources with correct IAM permissions can decrypt the data.

From this point forward, you are responsible for key rotation, correct IAM scoping and making sure that the application doesn’t leak any plain text information.

Architecture Overview

At a high level, the flow is simple:

User’s payload enter a Lambda function from API Gateway
Sensitive fields are encrypted immediately using AWS KMS
Encrypted data is written to DynamoDB

When data must be read and interpreted, the Lambda function explicitly decrypts the fields using the same KMS key from the encryption context.

Basically, there are 3 crucial operations:

Lambda calling KMS to encrypt and decrypt
Lambda reading and writing encrypted attributes in DynamoDB
IAM enforcing which principals are allowed to perform those operations

KMS Key Design in AWS CDK

In this pattern, the symmetric KMS key is the foundation. If its policy is overly permissive, field-level encryption becomes meaningless. The following CDK code is an example of creating a single customer-managed key dedicated for encryption and decryption processes for one service, with key rotation enabled and environment separation handled outside the snippet.

# Creation of a symmetric-key in KMS
pii_key = kms.Key(
    self,
    "PiiEncryptionKey",
    alias="pii-encryption-key",
    description="KMS key for encrypting PII data (latitude/longitude)",
    enable_key_rotation=True,
    removal_policy=RemovalPolicy.DESTROY,  # RETAIN for prod
)

At this layer, the key policy should be strict. Humans should not have decrypt permissions by default, and the Lambda execution role should be the primary and only consumer. CDK makes it easy to wire this correctly, but it also makes it easy to accidentally grant too much access if you are not careful enough.

Enabling Lambda IAM Access via Least Privilege

With the key in place, the next constraint is IAM. Encryption only works if decryption is tightly controlled. The Lambda execution role should have permission to encrypt and decrypt using exactly one KMS key and to read and write only the DynamoDB table it needs.

The following CDK code shows the Lambda function definition and the explicit grants for DynamoDB and KMS. There are no wildcards and no implicit permissions.

# CDK code for the Lambda + IAM Least Privilege Access
pii_handler = _lambda.Function(
    self,
    "PiiHandler",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="handler.handler",
    code=_lambda.Code.from_asset(
        os.path.join(os.path.dirname(__file__), "PIIHandler"),
    ),
    layers=[powertools_layer],
    timeout=Duration.seconds(15),
    memory_size=256,
    environment={
        "POWERTOOLS_SERVICE_NAME": "pii-service",
        "POWERTOOLS_LOG_LEVEL": "INFO",
        "PII_TABLE_NAME": pii_table.table_name,
        "KMS_KEY_ARN": pii_key.key_arn,
        "ALLOWED_ORIGIN": "*",
    }
)

# Enable the PIIHandler Lambda access to DynamoDB table
# and the KMS key to encrypt/decrypt the data
pii_table.grant_read_write_data(pii_handler)
pii_key.grant_encrypt_decrypt(pii_handler)

How to encrypt and decrypt data inside the AWS Lambda?

All encryption and decryption logic lives in the Lambda function, close to where data enters and leaves the system. The key rule is that plaintext PII should exist only in memory and only for as long as necessary.

The following Python Lambda code uses the KMS client to directly to encrypt and decrypt fields which are sensitive.

import json
import os
import base64

from boto3 import client, resource

# Environment variables, AWS Client and AWS resources
TABLE_NAME = os.environ.get("PII_TABLE_NAME")
KMS_KEY_ARN = os.environ.get("KMS_KEY_ARN")

KMS_CLIENT = client('kms')
DDB_RESOURCE = resource('dynamodb')
TABLE = DDB_RESOURCE.Table(TABLE_NAME)

def encrypt_data(plaintext: str) -> str:
        """Encrypt data using KMS and return base64-encoded ciphertext."""
    response = KMS_CLIENT.encrypt(
        KeyId=KMS_KEY_ARN,
        Plaintext=plaintext.encode("utf-8"),
        EncryptionContext={
            "purpose": "pii-location-encryption",
            "service": "pii-service",
        },
    )
    return base64.b64encode(response["CiphertextBlob"]).decode("utf-8")

def decrypt_data(encrypted_data: str) -> str:
        """Decrypt base64-encoded ciphertext using KMS."""
    ciphertext = base64.b64decode(encrypted_data)
    response = KMS_CLIENT.decrypt(
        CiphertextBlob=ciphertext,
        KeyId=KMS_KEY_ARN,
        EncryptionContext={
            "purpose": "pii-location-encryption",
            "service": "pii-service",
        },
    )
    return response["Plaintext"].decode("utf-8")

Writing Encrypted Data to DynamoDB

Encryption happens before saving data anywhere in the system and only the PIIHandler Lambda has access to encrypt and decrypt the data, meaning that DynamoDB never sees plaintext latitude or longitude values, only base64-encoded cipher text.

The following code snippet shows you the usage of the encrypt_data method I’ve shown above and the process of saving the sensitive information inside the DynamoDB table:

def handle_create_location(event) -> dict:
        """Handle POST /locations - Store encrypted PII location data."""
    event_body = json.loads(event.get('body'))

    user_id = event_body["user_id"]
    latitude = event_body["latitude"]
    longitude = event_body["longitude"]

    # ..

    # Encrypt PII fields (latitude and longitude)
    coordinates = json.dumps({"latitude": latitude, "longitude": longitude})
    encrypted_coordinates = encrypt_data(coordinates)

    item = {
        "pk": f"USER#{user_id}",
        "sk": f"LOCATION#{timestamp}",
        "encrypted_coordinates": encrypted_coordinates
    }

    TABLE.put_item(Item=item)

    # ...

AWS KMS Operational Considerations and Pricing

Now, to talk about some of the considerations. This approach does introduce some latency and cost because of the KMS calls, and fetching encrypted data inside the DynamoDB will increase item size overall. However, these are normal and predictable tradeoffs, which are rarely a limiting factor unless you have a system which is already operating at a very high scale.

More importantly, this pattern makes PII access, or any access to sensitive fields for that matter, explicit. Every decrypt operation is visible in CloudTrail, and access can be revoked centrally by modifying the KMS key policy. That property is far more valuable in production than marginal performance gains.

Regarding pricing, when writing this in February 2026, the price for one KMS key is $1USD per month (prorated hourly) and operations cost only $0.03USD per 10 000 requests, more information about pricing can be found by clicking on the link here. It’s worth mentioning that KMS operations are free for 20 000 operations in every AWS region which has KMS.

This pattern has key rotation enabled, meaning that every 365 days a new key will be generated and used for encrypting new data, which increases your cost per month by $1 USD. You don’t have to worry about the older data encrypted with the old key — KMS will save your older key which gets replaced and it will still be used to decrypt your data.

Conclusion

Field-level encryption in Lambda isn't advanced security — it should be a baseline for any system handling PII information.

The pattern above is made simple on purpose, but the constraints it enforces are fundamental: DynamoDB read access no longer implies an insight into sensitive data and any compliance audits shift straight to concrete log analysis, making your job easier to do.

As we’ve already discussed, the tradeoffs are real but manageable — KMS client calls add tens of milliseconds per operation and are relatively cheap (with a very generous Free tier as well!), meaning that this overhead is totally worth it because of the risk mitigation it provides.

The real question in a production environment isn’t whether encryption exists, but whether which computing resources have access to your KMS key and which resources are authorized to decrypt the sensitive data. By using this pattern, your future self (and your legal team if you have it in your company!) will thank you when the inevitable security questionnaire arrives.

Designing a Serverless Publishing Backend on AWS: Fan-Out as a First-Class Architecture Pattern

Matia Rašetina — Mon, 02 Mar 2026 08:00:00 +0000

Building a simple social platform where users can post about anything seems simple at first glance, until you reach the hardest problem: distributing content efficiently.

When you publish a blog post on any platform, did you ever wonder what happens next? For most of the platforms, many systems react to you publishing a post — content gets moderated, notifications about your activity sent, user’s feed updates. The real challenge here is doing everything without blocking the user or coupling the services together.

In this post, I’m going to tell you about fan-out architecture pattern — a strong pattern used to decouple services and enable data processing in multiple services at the same time.

This pattern will be used inside a full-stack blog post platform, like Medium, dev.to and others. The full-stack application will be covered in the next post.

Code for the whole backend is available by clicking on the link here.

Core AWS Services and Their Roles

At a high level, the system follows a simple rule:

APIs handle commands. Events handle consequences.

The backend is implemented using AWS CDK (Python) and relies exclusively on managed, serverless services. There are no always-on servers, no shared stateful services, and no synchronous dependency chains beyond the API boundary.

When a user publishes a post, the API does only three things:

Validate the request
Persist the post
Emit a Post Published event

Everything else — fetching the user’s feed, search indexing, notifications, post content moderation, analytics — happening in parallel and independent of each other.

This architecture uses the following serverless AWS services:

AWS Lambda (Python 3.12) for computing
AWS Cognito as a user management solution
Amazon DynamoDB as a database solution
Amazon SNS as the event bus (think of it as a "broadcast channel")
Amazon SQS for isolated, scalable workers
API Gateway for opening up the Lambdas to the Internet
AWS Bedrock for post content moderation via available LLMs

Here is the architecture diagram of the full-stack whole project, which we are going to take a closer look in the next blog post:

The Fan-Out Pattern: The Heart of the System

When a post is published, a single event is emitted to an SNS topic which sends the published post information to multiple SQS queues, as you can see in the architecture diagram.

Each downstream service subscribes via it’s own SQS queue, backed by a dedicated Lambda function worker.

This design has several important properties:

Loose coupling: Workers know nothing about each other and are completely independent
Fault isolation: One failure does not affect others
Independent scaling: Each worker scales based on its own load
Extensibility: New consumers of data can be added without changing existing code, just point to the new worker’s SQS queue

This is not just event-driven—it’s fan-out by construction, making it so easy to do multiple processes in parallel.

Another important detail to mention is that all SQS queue has it’s own Dead-letter queues (or DLQ for short), which collects the posts which couldn’t be processed, so the developer has an easier time debugging the failures in the system, making the system more mature and easier to work with. We are going to take a look how they work in the next blog post.

In the CDK code, creating a SNS topic and connecting the SQS queues looks like this:

self.post_published_topic = sns.Topic(self, "PostPublishedTopic")

def mk_queue(base: str) -> tuple[sqs.Queue, sqs.Queue]:
    dlq = sqs.Queue(self, f"{base}Dlq", retention_period=Duration.days(14))
    q = sqs.Queue(
        self,
        base,
        visibility_timeout=Duration.seconds(90),
        dead_letter_queue=sqs.DeadLetterQueue(queue=dlq, max_receive_count=5),
    )
    self.post_published_topic.add_subscription(subs.SqsSubscription(q))
    return q, dlq

self.feed_q, self.feed_dlq = mk_queue("FeedQueue")
self.search_q, self.search_dlq = mk_queue("SearchQueue")
self.email_q, self.email_dlq = mk_queue("EmailQueue")
self.moderation_q, self.moderation_dlq = mk_queue("ModerationQueue")
self.analytics_q, self.analytics_dlq = mk_queue("AnalyticsQueue")

Multiple Independent Workers Triggered by One Event

Every worker reacts to the same event but serves a different purpose:

Feed Worker: Distributes content to followers
Search Worker: Updates the inverted index
Email Worker: Sends notifications via SES
Moderation Worker: Runs AI moderation

Each worker:

Has its own IAM role
Has its own retry and DLQ configuration
Can fail without cascading impact

Connecting a Lambda worker to it’s SQS queue is done inside the CDK code like this:

# Any Lambda function, just copy, paste and adjust!
def mk_worker(name: str, path: str, queue, extra_env=None):
        fn = _lambda.Function(
            self,
            name,
            runtime=_lambda.Runtime.PYTHON_3_12,
            handler="handler.handler",
            code=_lambda.Code.from_asset(path),
            timeout=Duration.seconds(30),
            memory_size=256,
            environment={**common_env, **(extra_env or {})},
            layers=[powertools_layer],
        )
        fn.add_event_source(sources.SqsEventSource(queue, batch_size=10))

Now, let’s explain what each of the Lambda workers work in more detail.

Feed Worker

This Lambda takes the incoming data about the published post and updates the feed which should be shown to the followers of the user which published the post. A user’s feed is defined inside a DynamoDB table.

Moderation Worker

Content moderation is handled by a dedicated worker using AWS Bedrock’s Nova Micro model, because of it’s low price and it works wonders in this case.

Moderation is:

Asynchronous
Non-blocking
Cheap enough to run on every post

The worker takes the incoming data about the published post, like title, content and tags, builds a prompt with that data and sends it to the Nova Micro LLM model to analyze the content. The LLM will return either APPROVED or REJECTED, and the provided response is written inside the DynamoDB table, so the user cannot see any of the REJECTED posts.

Email Worker

This is very simple — take the incoming data of the blog post and send it to the followers found inside the DynamoDB table.

Search Worker

Instead of using OpenSearch or any other vector database, I’ve chosen to use DynamoDB for search indexing. The reason for choosing it is very simple — DynamoDB enables you to have the same functionality, but for a fraction of the cost.

All Lambda code is available in the Github repository available by clicking on the link at the beginning of this post.

Now that we’ve covered the heart of this application, there are some architectural decisions made for the DynamoDB data model and decisions made about exposing the API endpoints to the user.

Data Modeling: Designing for Access Patterns

The data layer is intentionally simple and optimized around how the application reads data, not how it looks relationally. This is explained as well, together with the fan-out pattern, as it’s very important for the application’s performance.

Posts Table

Stores the canonical representation of each post, indexed for:

Fetching a post by ID
Listing posts by author
Filtering by moderation status

Follows Table

Represents the social graph:

Partitioned by authorId
Enables fast lookup of followers when a post is published

Feed Table

This table exists to serve one purpose: fast feed reads.

Each user has their own partition containing pre-computed feed entries, sorted by publication time. There is no aggregation or computation at read time—fetching a feed is a single DynamoDB query.

This design only works because of one key decision: fan-out on write.

Search Index Table

Instead of using OpenSearch, the system implements a lightweight inverted index in DynamoDB by tokenizing titles and tags. It’s intentionally limited but dramatically cheaper and simpler for MVP-level search requirements.

The API Layer: Thin by Design

The API exposes a small set of endpoints:

Publish a post
Retrieve a post
Fetch a user’s feed
Search content

Very simple, very straight forward. To open up a Lambda to the Internet inside the CDK code, you would do it like this:

# Define the Lambda resource
get_post_fn = _lambda.Function(
    self,
    "GetPostFn",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="get_post.handler",
    code=_lambda.Code.from_asset("services/api/handlers"),
    timeout=Duration.seconds(10),
    memory_size=256,
    environment={
        "POSTS_TABLE": storage.posts_table.table_name,
        "POWERTOOLS_LOG_LEVEL": "INFO",
    },
    layers=[powertools_layer],
)

# Create the root API via API Gateway
http_api = apigw.HttpApi(
    self,
    "HttpApi",
    cors_preflight=cors_preflight,
)

# Add a route to the root API
http_api.add_routes(
    path="/posts/{postId}",
    methods=[apigw.HttpMethod.GET],
    integration=integrations.HttpLambdaIntegration("GetPostInt", get_post_fn),
)

Security and Identity

Authentication is handled by Amazon Cognito with:

Email-based signup
JWT-based authorization
OAuth 2.0 support

All APIs which are open to the Internet are protected via the Cognito authorizer, that way we let API Gateway handle protection of our endpoints.

In addition, each Lambda function inside this sytem operates under least-privilege IAM, with access only to the resources it requires. Workers or API Lambdas cannot accidentally read or modify unrelated data.

To protect the endpoint, you’ll need the created Cognito User Pool ID when it gets initialized. We’ll use the same Lambda we’ve opened up to the Internet via API Gateway, just modify the add_routes method arguments:

# Create Cognito authorizer
cognito_authorizer = apigw.HttpUserPoolAuthorizer(
    "CognitoAuthorizer",
    "ADD YOUR USER POOL ID HERE",
)

# Add a route to the root API
# and protect the endpoint with the Cognito Authorizer
http_api.add_routes(
    path="/posts/{postId}",
    methods=[apigw.HttpMethod.GET],
    integration=integrations.HttpLambdaIntegration("GetPostInt", get_post_fn),
    authorizer=cognito_authorizer
)

Cost Characteristics

All of services used in this project are inside the AWS Free tier and, if your traffic is low-to-medium (i.e. around few thousand users), most of the services are going to still be inside the Free tier. Only service which might cost you a couple of dollars will be AWS Bedrock, since running the project is running the LLM on every blog post.

Conclusion

You’ve now seen the power of this pattern and how easy it makes it to run multiple processes in parallel by using AWS serverless services.

The fan-out pattern turns a single publish action into a scalable pipeline of independent reactions. It keeps APIs fast, workers isolated, and the system easy to extend.

Most importantly, it aligns perfectly with the realities of a publishing platform:

Reads must be instant
Side effects must not block users
Features will grow over time and they are easy to add as SNS destinations

By making fan-out a first-class architectural choice, the backend remains simple—even as it scales.

In the next post, we’re going to take a closer look in all functionalities of the full-stack application.

How My AWS Lambda Runs Faster Than Yours: Here’s how to optimize Lambda Cold Starts with SnapStart

Matia Rašetina — Mon, 23 Feb 2026 09:04:41 +0000

If you ever worked with AWS Lambda, you know about one of their main weak points — cold starts. It’s been a pain point for any team building serverless applications with libraries that take lots of storage space. Whether it’s a data science processing library, machine learning models or complex custom libraries, cold starts can quickly become the dominant source of latency.

In early 2022, AWS launched Lambda SnapStart, a feature used to enable the developers to lower Lambda initialization times, and in late 2024, it was also enabled for Python runtimes as well. We’re going into more detail about SnapStart and how it works in the next heading.

In this post, I’ll walk you through how SnapStart works with Python 3.12, give you AWS CDK code examples which you can copy and paste straight into your project to enable SnapStart on your Lambdas, and show you how the impact of enabling SnapStart with real benchmark data from 10 cold start runs.

Full code can be found by clicking on the link here.

What Is AWS Lambda SnapStart?

Lambda SnapStart is a feature which enables the developers to mitigate the expensive and long Lambda environment initialization phases. What SnapStart does is it captures a snapshot of your execution environment when a new Lambda version is published and used for the first time, right after the initialization phase completes. For each subsequent cold start, the snapshot is restored as your Lambda execution environment instead of initializing the environment again, significantly reducing startup time.

For Python workloads, this is very impactful because if you are using Lambda for your ETL pipeline, smaller Machine Learning models, initializations of time-costly database connections etc., libraries like Pandas, NumPy and others are costing you precious time to initialize inside the Lambda environment and your user experience decreases dramatically. As mentioned, by using SnapStart and creating a snapshot of the Lambda environment, we are lowering the cold start time, which enables us to provide better user experience. When the Lambda is warm, the Lambda invocation time stays more or less the same, even if you have SnapStart enabled. The main point here is to only lower the cold start time.

Below you can see the request lifecycle with a standard Lambda execution environment and when SnapStart is enabled. This image is taken from the official AWS documentation, and if you want to read into more detail about SnapStart, you can do that by clicking on the link here.

Architecture Overview and Test Setup

The overview of this experiment is as following — we have 2 Lambdas, where one has SnapStart enabled and the other uses the standard Lambda environment without SnapStart.

We are only going to focus on importing the libraries in the Lambda environment, the Lambda handler method returns a “hello-world” back to the user, as we are not testing and measuring Lambda invocation times in this post. However, I’ve made the code log invocation time as well, so if you want to test something out in addition to importing the libraries, you can do that with ease.

Both have the same Lambda code — they are going to import some of the most used libraries like json, os, logging, however, then we are going to try and import libraries which cost Lambda initialization time the most — libraries like pandas, joblib and numpy.

Configuration of the Lambdas are:

ARM64 architecture
1024MB of memory
Python 3.12 used as Lambda runtime
30s timeout

Both Lambdas are run 10 times, with a 10 minute waiting time after any of the invocations, to have a very good chance of the Lambdas transition from a “warm” to a “cold” state. Information about the initialization times were taken from CloudWatch logs, and the Lambdas were invoked straight inside the AWS Console. Examples of the initialization times from CloudWatch can be seen in the following images:

Enabling SnapStart with AWS CDK (Python)

The following CDK code contains the project configuration — 2 AWS Lambdas, where one has SnapStart enabled, and the other one has SnapStart disabled, forcing it to have a standard Lambda execution environment.

from aws_cdk import (
  Stack,
  Duration,
  CfnOutput,
  aws_lambda as _lambda,
  aws_lambda_python_alpha as lambda_python,
  aws_iam as iam,
)
from constructs import Construct

class BenchmarkStack(Stack):
  def __init__(self, scope: Construct, id: str, **kwargs):
    super().__init__(scope, id, **kwargs)

        # 0. Creating a common configuration for each of the 2 Lambdas
    common = dict(
      runtime=_lambda.Runtime.PYTHON_3_12,
      entry="benchmark",
      index="handler.py",
      handler="handler",
      architecture=_lambda.Architecture.ARM_64,
      memory_size=1024,
      timeout=Duration.seconds(30),
      bundling=lambda_python.BundlingOptions(
        image=_lambda.Runtime.PYTHON_3_12.bundling_image,
        platform="linux/arm64",
      ),
    )

    # 1. SnapStart Lambda
    snapstart_fn = lambda_python.PythonFunction(
      self, "SnapStartFn",
      snap_start=_lambda.SnapStartConf.ON_PUBLISHED_VERSIONS, # Enable SnapStart for all published versions of the Lambda
      **common,
    )

    # 2. Standard Lambda (no SnapStart)
    standard_fn = lambda_python.PythonFunction(
      self, "StandardFn",
      # Do NOT include the "snap_start" argument, so it uses the regular
      # Lambda execution environment 
      **common,
    )

    # 3. Create versions
    snapstart_version = snapstart_fn.current_version
    standard_version = standard_fn.current_version

    # 4. Create aliases
    snapstart_alias = _lambda.Alias(
      self,
      "SnapStartAlias",
      alias_name="live",
      version=snapstart_version,
    )

    standard_alias = _lambda.Alias(
      self,
      "StandardAlias",
      alias_name="live",
      version=standard_version,
    )

It’s very important that the Lambda, which has SnapStart enabled, has an alias.

Lambda SnapStart requires aliases because it only works on published versions, and aliases provide a stable pointer to a specific version of your Lambda code with all of your requirements, code and libraries so AWS can safely create, cache, and restore the created snapshot.

Lambda Handler Code used for Testing Lambda SnapStart

# Light and quick imports of these libraries
import json
import logging
import os
import time
from typing import Any, Dict

# Heavy libraries which take time to initialize
# in the Lambda environment
import pandas
import numpy
import joblib

LOG = logging.getLogger()
LOG.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

INIT_EPOCH_MS = int(time.time() * 1000)
LOG.info(json.dumps({"phase": "init", "event": "complete", "ts_epoch_ms": INIT_EPOCH_MS}))

def handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    start = time.perf_counter()
    handler_start_epoch_ms = int(time.time() * 1000)

    # Lambda code for measuring total invocation time,
    # not the topic of this post
    # ...

As mentioned, to make the benchmark meaningful, the Lambda handler method includes heavy initialization logic, importing big libraries which we’ve mentioned above. This is to simulate a real-world Lambda doing a big computational task that loads large dependencies to complete it’s task.

Results: SnapStart vs No SnapStart (10 Cold Starts)

The difference between SnapStart-enabled and non-SnapStart Lambdas becomes immediately obvious when comparing cold start durations across multiple runs.

The average cold start initialization (so not total invocation, just initialization!) times were on average:

Lambda with a standard environment - 2929.99ms
Lambda with SnapStart enabled - 473.50ms

By only enabling one Lambda feature, we’ve cut the initialization time by ~84%! That’s 6.2x faster!

When SnapStart Is (and Isn’t) Worth It

We’ve seen how SnapStart makes a difference in cutting down on invocation times in Lambdas where library and global object initialization times are significant and can impact user experience.

It’s on you as a Lead Developer, or a Solutions Architect, to decide if implementing SnapStart in your Lambdas is worth it — if you are using AWS Lambda as your API handler to do CRUD operations in your application, more probably than not you are not going to need SnapStart enabled. Yet again, it’s on you to decide, based on the data you see.

Finally, there is some cost when using SnapStart (following cost is for eu-central-1, AWS region I’m using):

when caching the Lambda environment - $0.0000015046 per GB-second
when restoring the Lambda environment - $0.0001397998 for every GB restored

Cost is still very low, but something to keep in mind if you have Lambdas handling heavy traffic. More information about SnapStart pricing can be found on the link here.

Conclusion

AWS Lambda SnapStart offers a very powerful, yet a low-effort way of reducing cold start times for heavy serverless workloads. The benchmark provided in this blog post provides strong evidence that the Lambda functions, which require libraries that are time-consuming during Lambda environment initialization, can be mitigated by toggling on a single feature, making environment initialization times from a multi-second into sub-second starts.

For teams already using AWS CDK and Python Lambdas, enabling SnapStart is one of the highest-ROI optimizations available today.

CloudFront Signed Cookies Authorization: A Powerful Session-Based Pattern for Private Content on AWS

Matia Rašetina — Mon, 16 Feb 2026 08:00:00 +0000

Providing private content securely and swiftly is a very common challenge for any application which is on the internet. Whether you’re building a SaaS dashboard, a paid content platform or an enterprise portal, it is important that only authenticated users are able to access sensitive documents or assets, while providing best download performance.

This is where a Content Delivery Network service comes into play — on AWS, that’s CloudFront. CloudFront signed cookies authorization becomes a powerful design pattern. Instead of enforcing custom authorization logic on your backend, generating a pre-signed URL and sending it back to the user, here authentication happens in the application layer and authorization is handled at the edge by CloudFront.

By combining session-based authorization AWS patterns with edge security, you gain lower latency, improved scalability, caching and stronger security of your application. This article walks you through a clean, minimal, AWS-based architecture that shows you the thought process and the code to protect and provide private content to your users by using CloudFront signed cookies.

Code changes are in GitHub and are available by clicking on the link here.

High-Level Architecture Overview and AWS Services Used

At a high level, this pattern enables users to authenticate once via AWS Lambda and Cognito to then access application access with the best download performance. The browser or any other client of you application receives a HTTP-only cookie that CloudFront validates on every request, without calling any other backend services.

The core idea is simple:

Provide Authentication endpoints via AWS API Gateway and authenticate users using AWS Lambda and Amazon Cognito
Mint signed cookies via a lightweight backend
Enforce Amazon CloudFront authorization at the edge
Restrict the origin so content is never publicly accessible

Services which are crucial for this pattern are:

Amazon CloudFront + Amazon S3

Amazon CloudFront is a Content Delivery Network provided by AWS, which acts as the central enforcement point. It validates provided signed cookies and decides if the user has access to the assets available at the edge.

Amazon S3 is a service used to store objects and files inside buckets. The bucket blocks all public access and can only be reached through CloudFront, enabling CloudFront secure S3 access.

Amazon Cognito

Amazon Cognito handles user authentication. It verifies user identity and issues JWT tokens.

Amazon API Gateway + AWS Lambda

We use Amazon API Gateway to open up our Authentication Lambdas to the internet.

AWS Lambda runs a minimal, stateless function. Its single responsibility is to validate the incoming email and password and if the user is registered, it provides the signed cookies which enable the user to access private data. This keeps the system simple, auditable, and secure.

AWS Secrets Manager

AWS Secrets Manager securely stores the RSA private key used to sign CloudFront cookies. The key never leaves AWS-managed infrastructure.

End-to-End Request Flow and Architecture Diagram

The flow for the client of your application is the following:

User registers to the platform via Cognito and Register Lambda
User logs in and Login Lambda mints AWS CloudFront signed cookies
Browser stores cookies securely
When needed, browser requests private content from CloudFront
CloudFront validates the provided cookies
Authorized requests reach the private S3 bucket
Load the assets at lightning-fast speed!

Authentication Stack

First, we need to initialize our Cognito resources, like user pool, to be able to register and login users via our Lambdas, which we are going to define later. The following AWS CDK code creates the necessary resources in Cognito:

# Cognito User Pool
user_pool = cognito.UserPool(
    self,
    "UserPool",
    user_pool_name="UserPool",
    auto_verify=cognito.AutoVerifiedAttrs(email=True),
    sign_in_aliases=cognito.SignInAliases(email=True),
    self_sign_up_enabled=True,
    password_policy=cognito.PasswordPolicy(
        min_length=8,
        require_lowercase=True,
        require_uppercase=True,
        require_digits=True,
        require_symbols=False,
    ),
    standard_attributes=cognito.StandardAttributes(
        email=cognito.StandardAttribute(required=True, mutable=True),
    ),
    removal_policy=RemovalPolicy.DESTROY,
)

user_pool_client = cognito.UserPoolClient(
    self,
    "UserPoolClient",
    user_pool=user_pool,
    user_pool_client_name="WebApp",
    generate_secret=False,
    auth_flows=cognito.AuthFlow(
        user_srp=True,
        user_password=True,
        custom=False,
        admin_user_password=False,
    ),
    prevent_user_existence_errors=True,
)

Generating and Saving the RSA Key-pair

To enable the signing process of the cookies, we need to generate the RSA key-pair, create a variable that will hold the public key and make it eligible for usage with CloudFront and save the private key in the AWS Secrets Manager.

Important note — this pattern creates the RSA key-pair every time you try to deploy or update the stack. This is made on purpose, as the pattern itself is made to get you quickly off the ground. The best way is to generate the keys in a separate Python script or via your terminal, load them in the CDK code and deploy to AWS.

# Generate private key
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
    backend=default_backend()
)

# Serialize private key to PEM format
private_key_pem = private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.TraditionalOpenSSL,
    encryption_algorithm=serialization.NoEncryption()
).decode('utf-8')

# Extract public key and serialize to PEM format
public_key = private_key.public_key()
public_key_pem = public_key.public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')

cf_public_key = cloudfront.PublicKey(
    self,
    "CloudFrontPublicKey",
    encoded_key=public_key_pem,
    comment="Public key for CloudFront signed cookies",
)

cf_key_group = cloudfront.KeyGroup(
    self,
    "CloudFrontKeyGroup",
    items=[cf_public_key],
    comment="Key group for signed cookie validation",
)

# Secrets Manager - Private Key Storage
private_key_secret = secretsmanager.Secret(
    self,
    "CloudFrontPrivateKeySecret",
    description="CloudFront private key for signing cookies (PEM format)",
    secret_string_value=SecretValue.unsafe_plain_text(private_key_pem),
    removal_policy=RemovalPolicy.DESTROY,
)

Now, that we have our encryption keys and all variables initialized, now it’s time to create the CloudFront distribution itself.

Creating S3 bucket, CloudFront Distribution and Securing the S3 Bucket

The S3 bucket is fully locked down:

All public access is blocked
No direct access from the internet is allowed
Requests are accepted only from CloudFront

This is achieved using Origin Access Control, enabling AWS private S3 bucket CloudFront OAC patterns. Even if someone discovers the S3 bucket name, they cannot retrieve objects directly.

The CloudFront key group created in the previous step is referenced here to associate the public key from our key pair with the newly created distribution, allowing CloudFront to recognize and use it correctly.

In addition, it’s worth mentioning that assets and object that have the prefix private/ in their key are protected via the signed cookie mechanism, while all other assets in the bucket are publicly available via the CloudFront distribution.

This design ensures your private content delivery AWS strategy has no weak points.

# S3 Bucket for Private Assets
private_assets_bucket = s3.Bucket(
    self,
    "PrivateAssetsBucket",
    block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
    encryption=s3.BucketEncryption.S3_MANAGED,
    enforce_ssl=True,
    removal_policy=RemovalPolicy.DESTROY,
    auto_delete_objects=True,
    versioned=False,
)

# CloudFront Origin Access Control (OAC)
oac = cloudfront.CfnOriginAccessControl(
    self,
    "S3OriginAccessControl",
    origin_access_control_config=cloudfront.CfnOriginAccessControl.OriginAccessControlConfigProperty(
        name=f"{construct_id}-S3-OAC",
        origin_access_control_origin_type="s3",
        signing_behavior="always",
        signing_protocol="sigv4",
        description="OAC for private S3 bucket access via CloudFront",
    ),
)

# CloudFront Distribution
# Create the S3 origin without OAC first
s3_origin = origins.S3BucketOrigin.with_origin_access_control(
    private_assets_bucket
)

# Default behavior - public content (no signed cookies required)
default_behavior = cloudfront.BehaviorOptions(
    origin=s3_origin,
    viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
    allowed_methods=cloudfront.AllowedMethods.ALLOW_GET_HEAD,
    cached_methods=cloudfront.CachedMethods.CACHE_GET_HEAD,
    cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED,
    compress=True,
)

# Private behavior - requires signed cookies
private_behavior = cloudfront.BehaviorOptions(
    origin=s3_origin,
    viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
    allowed_methods=cloudfront.AllowedMethods.ALLOW_GET_HEAD,
    cached_methods=cloudfront.CachedMethods.CACHE_GET_HEAD,
    cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED,
    compress=True,
    trusted_key_groups=[cf_key_group],
)

distribution = cloudfront.Distribution(
    self,
    "PrivateContentDistribution",
    default_behavior=default_behavior,
    additional_behaviors={
        "private/*": private_behavior,
    },
    comment="Distribution for private S3 content with signed cookies",
    price_class=cloudfront.PriceClass.PRICE_CLASS_100,
    enabled=True,
)

# S3 Bucket Policy for CloudFront OAC
private_assets_bucket.add_to_resource_policy(
    iam.PolicyStatement(
        sid="AllowCloudFrontServicePrincipalReadOnly",
        effect=iam.Effect.ALLOW,
        principals=[iam.ServicePrincipal("cloudfront.amazonaws.com")],
        actions=["s3:GetObject"],
        resources=[private_assets_bucket.arn_for_objects("*")],
        conditions={
            "StringEquals": {
                "AWS:SourceArn": f"arn:aws:cloudfront::{self.account}:distribution/{distribution.distribution_id}"
            }
        },
    )
)

AWS Lambdas + API Gateway

Finally, we have our 2 Lambdas and API Gateway which enables the user to register and login into our platform to receive the necessary credentials, like the signed cookie and idToken in case you are going to use Cognito Authorizer on your other stacks.

# Lambdas
# IAM Policy for Lambda functions to access Cognito
cognito_policy = iam.PolicyStatement(
    effect=iam.Effect.ALLOW,
    actions=[
        "cognito-idp:SignUp",
        "cognito-idp:AdminConfirmSignUp",
        "cognito-idp:InitiateAuth",
    ],
    resources=[user_pool.user_pool_arn],
)

# Register User Lambda Function
register_lambda = _lambda.Function(
    self,
    "RegisterUserLambda",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="lambda_handler.handler",
    code=_lambda.Code.from_asset(
        "./lambda/Register",
        bundling={
            "image": _lambda.Runtime.PYTHON_3_12.bundling_image,
            "command": ["bash", "-c", "pip install aws-lambda-powertools -t /asset-output && cp -r . /asset-output"],
        },
    ),
    environment={
        "POWERTOOLS_SERVICE_NAME": "authentication",
        "USER_POOL_ID": user_pool.user_pool_id,
        "USER_POOL_CLIENT_ID": user_pool_client.user_pool_client_id,
        "ALLOWED_ORIGIN": allowed_cors_origin,
    },
    timeout=Duration.seconds(30),
)
register_lambda.add_to_role_policy(cognito_policy)

# Login User Lambda Function
login_lambda = _lambda.Function(
    self,
    "LoginUserLambda",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="lambda_handler.handler",
    code=_lambda.Code.from_asset(
        "./lambda/Login",
        bundling={
            "image": _lambda.Runtime.PYTHON_3_12.bundling_image,
            "command": [
                "bash",
                "-lc",
                "pip install --platform manylinux2014_x86_64 --only-binary=:all: --no-cache-dir --upgrade "
                "-t /asset-output aws-lambda-powertools 'cryptography>=41' "
                "&& cp -au . /asset-output"
            ],
        },

    ),
    environment={
        "POWERTOOLS_SERVICE_NAME": "authentication",
        "USER_POOL_ID": user_pool.user_pool_id,
        "USER_POOL_CLIENT_ID": user_pool_client.user_pool_client_id,
        "ALLOWED_ORIGIN": allowed_cors_origin,
        "PRIVATE_KEY_SECRET_ARN": private_key_secret.secret_arn,
        "CLOUDFRONT_DOMAIN": distribution.distribution_domain_name,
        "KEY_PAIR_ID": cf_public_key.public_key_id,
        "COOKIE_TTL_SECONDS": str(cookie_ttl_seconds),
        "COOKIE_DOMAIN": cookie_domain,
        "COOKIE_SAME_SITE": same_site,
    },
    timeout=Duration.seconds(30),
    memory_size=256,
)
login_lambda.add_to_role_policy(cognito_policy)
private_key_secret.grant_read(login_lambda)

# API Gateway REST API
api = apigw.RestApi(
    self,
    "AuthApi",
    rest_api_name="Auth and Cookie API",
    description="API for authentication and CloudFront signed cookies",
    deploy=True,
    deploy_options=apigw.StageOptions(stage_name="v1"),
    default_cors_preflight_options=apigw.CorsOptions(
        allow_origins=[allowed_cors_origin] if allowed_cors_origin != "*" else apigw.Cors.ALL_ORIGINS,
        allow_methods=["POST", "OPTIONS"],
        allow_headers=["Content-Type", "Authorization", "X-Amz-Date", "X-Api-Key"],
        allow_credentials=True,
    ),
)

# API Gateway Lambda Integration
register_integration = apigw.LambdaIntegration(register_lambda)
login_integration = apigw.LambdaIntegration(login_lambda)

# API Gateway Resources and Methods
api.root.add_resource("register").add_method("POST", register_integration)
api.root.add_resource("login").add_method("POST", login_integration)

After successful registration via the Register Lambda, when the user tries to login, they invoke the Login Lambda. This function is intentionally minimal:

Validate the authenticated request context
Retrieve the RSA private key from Secrets Manager
Generate CloudFront signed cookies architecture values
Return the cookies to the browser as HTTP-only cookies

Because the cookies are HTTP-only, they cannot be accessed by JavaScript, reducing the risk of XSS attacks. This is a critical best practice when implementing secure content delivery using CloudFront.

Security Best Practices

To harden your CloudFront signed cookies authorization setup:

Use short-lived cookies
Rotate RSA keys regularly
Grant least-privilege IAM permissions
Enable logging and monitoring at CloudFront and S3

These practices strengthen trust and align with AWS security recommendations.

By default, the cookie will expire in 600 seconds or 10 minutes after it’s created. That’s something you can modify inside the CDK code on your own, and also you can configure CORS settings as well.

Testing the pattern

I’ve uploaded 2 images from the AWS console into the S3 bucket - one is a picture of a car and it’s S3 key is private/gt3.jpg , while the other image is of Manhattan in New York City and it’s key is public/Above_gotham.jpg .

We are going to do 3 tests:

try and fetch the private/gt3.jpg image without the signed cookie - we expect this request to be blocked
try and fetch the same image but with the signed cookie - we expect this to pass
try and fetch the image of Manhattan, which we expect to pass since it’s a publicly available asset via the CloudFront distribution URL

I’ve used Postman to test this out and here is the example of the failed request, where the signed cookie is required:

Now, we are going to sign in, get the cookie, put it in the Cookie header and send the request again:

Finally, let’s try and fetch the publicly available image:

This is proof that our pattern works as expected — all assets and objects which are inside the private/ folder in S3 are protected and users need to be signed in to access them, while some objects can still be accessed publicly if needed.

When This Pattern May Not Be the Right Fit

While CloudFront signed cookies offer very powerful benefits, they also introduce tradeoffs which are worth considering.

This approach works best for content which doesn’t change often and has predictable access patterns, like videos, images and documents. For highly dynamic content, that changes per-request and/or requires real-time authorization, the edge-based model becomes limiting since you cannot revoke the cookie session in progress without cookie expiration, as it’s valid until it expires.

Additionally, this pattern adds architectural complexity — you need to manage RSA key pairs, coordinate cookie domains and if there’s a bug, you have multiple AWS services to look into.

If you have a smaller application and don’t really mind the longer download times and want less complex architecture for your application, a simpler approach would be to have a Lambda which handles pre-signing process of the S3 objects, as that is simpler and easier to maintain.

Conclusion

This architecture demonstrates how authentication happens by using just a few AWS services, while authorization is enforced at the edge by CloudFront. By combining Cognito, API Gateway, Lambda, and CloudFront signed cookies, you create a robust, scalable, and secure system for private content delivery AWS.

If you’re designing modern cloud systems, adopting CloudFront signed cookies authorization is a practical step toward better performance and stronger security.

Redis vs DynamoDB vs DAX: I Benchmarked AWS Caching Performance (The Results Were Unexpected)

Matia Rašetina — Mon, 02 Feb 2026 08:00:00 +0000

In many backend systems, user data is fetched on almost every request. A common assumption is that adding an in-memory cache will improve read performance by any system.

To validate this assumption, I’ve benchmarked 3 approaches of accessing user data from DynamoDB inside the Serverless infrastructure:

A baseline using AWS Lambda and DynamoDB
A cache-aside approach using Redis
AWS Lambda backed by DynamoDB Accelerator (DAX)

I wanted to find out, if you are a CTO of a startup and you are thinking of implementing caching services into your architecture, to see if it’s worth. I expected that the cached approaches will outperform the baseline easily.

Even at 200 requests per second (that’s 12 000 requests per minute!), that assumption didn’t hold.

Experimental Setup

The architecture across the benchmark is as identical as possible — the cheapest available option on AWS region eu-central-1.

Here are the 3 approaches

Baseline (DDB + Lambda)

Lambda: Python 3.12, 256 MB memory, 10s timeout
DynamoDB: Pay-per-request billing mode
Access Pattern: Direct reads from DynamoDB without any caching layer
Latency: Highest (no cache)

DAX (DynamoDB Accelerator)

Lambda: Python 3.12, 256 MB memory, 10s timeout
DAX Cluster:
- Node type: dax.t3.small
- Replication factor: 1 (single node)
- Deployed in VPC isolated subnets
Cache: Managed by DAX automatically (default item TTL ~5 minutes, query cache TTL ~5 minutes)
Access Pattern: Lambda → DAX → DynamoDB
Client: Uses amazondax Python client available on PIP

Redis (AWS ElastiCache)

Lambda: Python 3.12, 256 MB memory, 10s timeout
Redis Cluster:
- Node type: cache.t4g.micro
- Single node (no automatic failover)
- Deployed in VPC isolated subnets
Cache TTL: 30 seconds (configurable via REDIS_TTL_SECONDS)
Access Pattern:
- Lambda checks Redis first
- On miss: reads from DynamoDB, stores in Redis with 30s TTL
- On hit: returns cached data directly
Client: Standard redis-py client with 1s connection timeout

I’ve tested 2 load levels — one with 50 reads per second (3000 reads per minute), the other 200 reads per second (12000 reads per minute), by using the same payload size, hot/mixed key distribution, and p95 latency as the primary metric. Testing was done with an open-source library k6 with Javascript.

In all 3 approaches, the same data was fetched from the database. Here is an example:

{
  "pk": "ITEM#123",
  "sk": "META",
  "itemId": "123",
  "title": "Example Item Title",
  "body": "This is the body content of the item",
  "updatedAt": 1736467200,
  "etag": "a7f8d9e1c2b3a4f5e6d7c8b9a0f1e2d3c4b5a6f7e8d9c0b1a2f3e4d5c6b7a8f9"
}

Results at 50 RPS — Establish the baseline truth

Access Pattern	p95 Latency (ms)	Avg Latency (ms)	Dropped Iterations	Notes
Lambda + DynamoDB (Baseline)	~63 ms	~48 ms	0	Fast, stable, no bottlenecks
Redis (warmup run)	~68 ms	~66 ms	22	Cache misses + write-back cost
Redis (steady state)	~63 ms	~48 ms	0	Matches baseline, no latency win
DAX (single small node)	~1040 ms	~957 ms	19	Cache saturation, unusable

Baseline: Lambda + DynamoDB @ 50 RPS

At 50 RPS, the most common and baseline setup by using Lambda + DynamoDB achieved a p95 latency of ~63ms, with no dropped requests. Hot and mixed key access patterns had very similar latency, which indicates that DynamoDB on-demand wasn’t under pressure at all.

Redis: Warmup vs Steady State @ 50 RPS

The Redis results clearly show two very different phases — during the first run, the p95 latency was, on average 66ms. This is expected behavior, as the system asks the Redis cluster if the requested data is already inside it. If it’s not, the Lambda proceeds to query the data inside the DynamoDB. Since the cluster doesn’t have any data cached, this latency is expected.

Once the cache was warm, Redis did it’s job and achieved nearly identical latency to the baseline. However, it did not outperform it.

DAX: Undersized Cache Failure @ 50 RPS

The DAX configuration performed significantly worse than both the baseline and Redis. With a single small node, the DAX cluster became CPU-bound, leading to request queueing and p95 latencies exceeding one second.

This result highlights an often-overlooked risk: a misconfigured or undersized cache can actively degrade performance. DAX is not a drop-in optimization — it requires careful capacity planning. Keep in mind, we are running the benchmark with the weakest instance of DAX available.

Conclusion: 50 RPS

At 50 RPS, the baseline configuration performed great and doesn’t require any additional caching, as it only introduces additional network and service hops, while getting not much in return. Yes, by using Redis, DynamoDB pressure went down, but still wasn’t even near it’s limit.

Results at 200 RPS — The expected crossover that didn’t happen

Access Pattern	p95 Latency (ms)	Avg Latency (ms)	Dropped Iterations	Notes
Lambda + DynamoDB (Baseline)	~63 ms	~48 ms	13	Stable, scales linearly
Redis (warm run)	~64 ms	~52 ms	40	Cache population under load
Redis (steady state)	~70 ms	~58 ms	79	Slightly worse than baseline
DAX (single small node)	~1050 ms	~968 ms	5,399	Cluster saturation

Baseline: Lambda + DynamoDB @ 200 RPS

At 200 requests per second, the baseline Lambda + DynamoDB configuration had a nearly identical performance to the 50 RPS run. p95 latency remained around 63 ms, with only a small number of dropped iterations.

This makes a very strong case that the baseline is a very optimized choice by default.

Redis @ 200 RPS: Warm vs Steady State

As with the 50 RPS test, Redis exhibited two distinct phases. During the initial run, cache misses caused additional latency and tail spikes as items were fetched from DynamoDB and written back to Redis.

Once the cache was warm, Redis stabilized but did not outperform the baseline. In steady state, p95 latency increased slightly to ~70 ms, and the number of dropped iterations was higher than with DynamoDB alone.

DAX @ 200 RPS: Saturation Under Load

At 200 RPS, the DAX configuration began to collapse. Effective throughput dropped well below the target rate, p95 latency exceeded one second, and thousands of iterations were dropped.

This behavior confirms that DAX is highly sensitive to sizing — smaller instances simply do not provide any benefit.

Conclusion: 200RPS

Even at 200 RPS, the dominant cost in this system was not database access but network and managed service overhead. Adding a cache did not remove that cost — it added to it, as you still need to pay for the on-demand or serverless Redis instance, depending on your choice.

What These Results Actually Prove

With these benchmarks, we’ve proven:

DynamoDB on-demand scales extremely well for simple reads
Redis reduces pressure, not latency
Cache warmup matters
Misconfigured DAX is worse than no cache
Latency optimization and scaling optimization are different problems

Conclusion & Lessons Learned

The results from both the 50 RPS and 200 RPS benchmarks lead to a clear and somewhat counterintuitive conclusion: for this workload, Lambda backed by DynamoDB on-demand was already fast enough that adding a cache did not improve user-visible latency.

At both load levels, DynamoDB was not the bottleneck. End-to-end latency was dominated by network distance and managed service overhead rather than database access time. As a result, introducing Redis added an extra network hop and client-side overhead without removing the dominant cost in the request path.

Redis still served a purpose, but not the one I initially expected. It reduced pressure on DynamoDB and flattened backend load, which can be valuable for cost control and future scaling. What it did not do — at least at these traffic levels — was make requests faster.

DynamoDB Accelerator, based on these results, taught us a totally different lesson. When undersized, it simply doesn’t help in anyway, as it got saturated very quickly which caused increased latency and dropped requests. It requires careful capacity planning and workflow understanding.

And maybe the biggest lesson while doing this experiment is the understanding that doing a measurement run before optimizing is incredibly important. Even though it makes sense to cache the data to save the resources and ease the load on the backend, it doesn’t cause the end-to-end latency to lower.

It seems that this workflow is just too simple to see the caching benefits — if you have a time-costly method inside your code and the data can be cached, most probably caching would be worth testing.

In short: don’t cache because it feels right—cache because the data proves you need it.

Running a RAG Pipeline in a Production Full-Stack Application (Without a Vector Database)

Matia Rašetina — Mon, 19 Jan 2026 08:00:00 +0000

In the previous post, I focused on how to build a Retrieval-Augmented Generation (RAG) pipeline on AWS using DynamoDB as a low-cost vector store. The goal there was simple: prove that you don’t need heavy infrastructure or a dedicated vector database to get something useful working.

If you haven’t read it, you can check it out by clicking on the link here.

This post picks up where that one stops. We will implement the RAG pipeline we’ve created into a frontend application, which uses our backend and creates a full-stack application, which you can use to upload documents and get easier access to information you need from your PDFs. We will see how the budget RAG pipeline behaves with real users and traffic.

The link to the GitHub repository is the same as from the previous post, it’s just located on a branch called full-stack-implementation — access the repo by clicking here.

What the Application Does

The frontend application is written with the following technologies:

Next.js - React framework with App Router for server-side rendering and routing
Tailwind CSS 4 - Utility-first CSS framework for styling
shadcn/ui - Component library built on Radix UI primitives (New York style)
NextAuth.js - Authentication solution for Next.js
Lucide React - Icon library

When you load up the application, the user is prompted with a signup and a login form, which uses the Authentication stack on the backend.

The Authentication stack uses AWS Cognito for easier user management.

After the user signs up and logs in, they can upload PDF documents to the platform, wait for them to be indexed, and then ask questions that are answered using only the content of those documents.

There’s no attempt to hide latency, no background magic, and no assumption of massive scale. This is designed for early-stage usage: experimentation, internal tools, or MVPs where validating the feature matters more than optimizing for peak performance.

I’m going to show you 2 happy paths, one for the document upload, and the other for asking questions.

Here is a GIF which shows you how the user can upload the documents to the platform. I’ve exported my old blog posts as PDFs and uploaded them to the platform, just as an example.

After the documents got indexed by the backend and saved to the DynamoDB table, they will come up inside the Chat UI. You can choose which documents you want to ask questions about — the backend will take the ID of the uploaded document, get the information from the database and get the answer to your question with Bedrock LLM.

High-Level Architecture

The simple frontend talks to an API layer backed by AWS Lambda.

There are 2 stacks, one for authentication and user management, the other is the stack for the documents. Separating Authentication and Document processing keeps the ingestion and query path independent. In addition, if you ever wanted to expand this application and the other stack needs to have it’s API Gateway protected, you can simply import the Authentication stack’s User Pool ID and easily protect your endpoints.

As mentioned, authentication and user management is completely handled by AWS Cognito, together with a Login and Register Lambdas which handle and validate the incoming payloads.

Regarding the Document stack, documents are stored in S3 via the presigned URLs, embeddings and metadata live in DynamoDB, and the LLM and embedding models are accessed via AWS Bedrock. There are no long-running services, ensuring that there are no fixed monthly costs.

Every architectural choice here optimizes for one thing: minimizing cost and operational overhead, while still providing the RAG pipeline development experience and it’s features.

API and Lambda Performance with Real Traffic

The following image shows you the execution duration of the Lambda which indexes the incoming documents and puts the information inside the DynamoDB table. Remember the architecture from the previous post, when the document is uploaded in S3, it triggers an EventBridge rule which puts the uploaded document information into SQS queue, which triggers the document indexing Lambda. This is done like this to reuse Lambda environments and to lower Lambda invocations, lowering cost.

The first points are Lambda invocations where 1 PDF of 400kb is uploaded, later points are Lambda invocations where 2 PDFs are uploaded, both have ~200kb.

Indexing two small PDFs in just over 8 seconds means document indexing stays comfortably asynchronous. Users aren’t blocked, and indexing cost remains predictable.

Next, you can list indexed documents, inspect their state, and see exactly when they become available for querying. In the meantime, I’ve uploaded the same documents and other blog PDFs with similar sizes and reached around 20+ documents in the database and got very similar results, around ~4 seconds per document to index it and put it inside the database.

Now, we to fetch all available documents for the user, we need to call the get-documents API. The screenshots below show the get-documents endpoint in action, along with the execution duration of the document indexing Lambda. For testing, I’ve used Postman API testing and got the following results when querying all documents inside the database for my user. Notice that the first API call is slower than the rest? That’s because of the Lambda cold start. It’s important to note that an average user of the application won’t experience cold starts often, especially if there are other users on the platform.

Finally, the Ask Questions Lambda which the user invokes when they choose the documents about which they want to send an inquiry about. It’s a simple payload to the Lambda, where the selected document IDs are sent, together with a question.

The following graph is the duration of the execution of the said Lambda. The Lambda was tested first to see the cold start duration, then the warm start duration. First couple of executions questions regarding only one document — the question was:

What is this document about?

Whilst the latest execution in the graph had 4 documents chosen with the question:

Give me a summary of these 4 documents.

As you can see, when inquiring about one document, the cold start duration is around the 3 second mark, while when the Lambda is warm, the execution time is under 1.5 seconds.

On the other hand, the cold start duration when the Lambda is asked about 4 documents is just above 4 seconds, while the warm start duration is just under 4 seconds.

As mentioned before, an average user won’t experience many Lambda cold starts. For a budget RAG system, using very few AWS services for minimal complexity, these performances are acceptable.

Conclusion

This application confirms what the first blog post theorized — a DynamoDB-based RAG system is not just cheap, but usable and if you understand it’s limits, has acceptable performance.

As we’ve discussed in the previous post, trade-offs made are obvious and expected — Lambda cold starts are real, document query latency behaves as expected and increases if the document count grows.

For early-stage applications, internal tools, experiments or even a hackathon competition where you don’t want to break the bank, but still have a working RAG system, this setup does its job well. It lets you ship the product, let your users give you feedback on the application and delay expensive architectural decisions until data forces your hand.

You Don’t Need a Vector Database to Build RAG (Yet): A ~$1/Month DynamoDB Pipeline

Matia Rašetina — Mon, 12 Jan 2026 08:00:00 +0000

You've just been tasked with building a document Q&A system, where the users are able to upload PDF documents and then ask questions to get accurate answers based on the content in the uploaded documents. This request is the perfect opportunity to build a Retrieval Augmented Generation (RAG) pipeline. For starters, you’ve been asked to do a Proof-of-concept project therefore, you don’t have a budget to create this project with best tools available for this scenario.

If you’ve never researched how a RAG system works, it works in the following way - it breaks down a document into multiple chunks, converts them into a vector embeddings (numerical representations of the incoming data), storing them in a database and then later, when the system is asked about the document, it finds the most relevant chunks inside the database. The results are then fed into a Large Language Model, which then provides an answer.

I wanted to learn about RAG pipelines as cheap as possible, without using a vector database, even though I know that’s a better option performance-wise.

That’s why I decided to use DynamoDB. This post will show you that it’s one of the cheapest implementation of a RAG pipeline available. Of course, there are drawbacks and trade-offs, but for a strict budget and POC pipeline, it’s a great choice.

Technologies and services which we’ll be using are:

Python and AWS CDK in Python
AWS Bedrock
AWS Lambda
AWS DynamoDB
AWS SQS

Link to the GitHub repository is here - https://github.com/mate329/budget-rag-system-with-dynamodb.

Let's dive in.

Architecture Overview

Take a look into the architecture of the pipeline, together with explained steps:

Document Upload: The user requests a pre-signed S3 URL from the Upload Handler Lambda to upload the document into the S3 bucket
Event-Driven Ingestion: Upon document upload in S3, it triggers an EventBridge rule that sends the event to an SQS queue, where the messages are batched to reduce Lambda invocations and cuts costs significantly
Document Processing: The Index Document Lamdba is triggered by SQS batches. For each document, it:
- Downloads the file from S3 via the provided S3 key in the payload
- Splits the document into semantic chunks
- Generates vector embeddings using Bedrock (Titan Embeddings or similar)
- Stores chunks + embeddings + metadata in DynamoDB
Question Answering: When a user asks a question via the Ask Question Lambda:
- The question is converted to a vector embedding (same model as step 3)
- The database is searched for the most similar chunks
- Top N relevant chunks are retrieved as context and provided to a Bedrock LLM (in this project, we use Amazon’s Titan model to keep the cost down) to generate an answer
- The answer is returned to the user

DynamoDB as a Vector Store

When I started building this RAG system, I had two goals: ship fast and keep costs as low as possible.

Here's the mental shift required: DynamoDB isn't a vector database, and that's fine. It doesn't have native k-NN search or approximate nearest neighbor algorithms. But it does have:

Single-digit millisecond latency for queries
Built-in user isolation through partition keys
Zero operational overhead (no cluster management, no indexing jobs)
Cost that scales to zero when not in use

For an MVP with 10-100 documents and a handful of users? DynamoDB is perfect. You're trading query performance for operational simplicity and cost savings. That's a trade worth making early on.

Architecture & Data Model

Defining the table inside the CDK is easy:

# 3. Create DynamoDB table with USER ISOLATION
self.vectors_table = dynamodb.Table(
    self, "DocumentVectorsTable",
    table_name="document-vectors",
    partition_key=dynamodb.Attribute(
        name="user_id",
        type=dynamodb.AttributeType.STRING
    ),
    sort_key=dynamodb.Attribute(
        name="document_page"
        type=dynamodb.AttributeType.STRING
    ),
    billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
    removal_policy=RemovalPolicy.DESTROY
)

# GSI for listing user's documents
self.vectors_table.add_global_secondary_index(
    index_name="user-document-index",
    partition_key=dynamodb.Attribute(
        name="user_id",
        type=dynamodb.AttributeType.STRING
    ),
    sort_key=dynamodb.Attribute(
        name="document_id",
        type=dynamodb.AttributeType.STRING
    ),
    projection_type=dynamodb.ProjectionType.KEYS_ONLY
)

The key to making DynamoDB work as a vector store is the table design:

# Partition Key: user_id
# Sort Key: document_page (format: "doc_id#page_num")
{
    'user_id': 'user-123', # user_id is derived from Cognito user ID
    'document_page': 'quarterly-report-2024#0',  # doc_id#page_num
    'document_id': 'quarterly-report-2024',
    'page_number': 0,
    'page_text': 'Executive Summary...',
    'page_vector': [0.123, -0.456, 0.789, ...],  # 1536 dimensions from Titan
    's3_bucket': 'my-documents',
    's3_key': 'user-123/quarterly-report-2024.pdf'
}

Why this design?

User isolation is automatic – Querying with user_id as the partition key inside the Ask Questinon Lambda means User 1 can never see User 2's data. No filter logic, no mistakes.
Composite sort key allows efficient queries like "get all pages of this document"
Vectors as native lists – DynamoDB supports List types, no serialization needed

The GSI on user_id + document_id lets users list their documents without scanning all pages.

PDF Document Indexing Flow

Before the Index Lambda processes the incoming documents, all S3 notifications about uploaded documents to the bucket get put into the SQS queue. All of the notifications are batched and this is where SQS batching is crucial. Without it, you'd pay for one Lambda invocation per document. With batching, 10 documents = 1 invocation = 90% cost savings on Lambda.

Here's the document processing code for each of the incoming PDFs, where:

we take each page of the incoming PDF document and check if it has text data on it
communicate with the Bedrock model to embed the data
save the embedded data into the DynamoDB table

def get_embedding(text: str) -> list:
    """Generate embedding for text using Bedrock."""
    try:
        response = bedrock_runtime.invoke_model(
            modelId='amazon.titan-embed-text-v1',
            body=json.dumps({"inputText": text})
        )

        response_body = json.loads(response['body'].read())
        return response_body.get('embedding')
    except Exception as e:
        print(f"Error generating embedding: {str(e)}")
        raise

def process_pdf(bucket: str, key: str, document_id: str, user_id: str):
    """Process PDF and store page vectors in DynamoDB."""
    table_name = os.environ['DYNAMODB_TABLE_NAME']
    table = dynamodb.Table(table_name)

    # Download PDF from S3
    pdf_obj = s3_client.get_object(Bucket=bucket, Key=key)
    pdf_data = pdf_obj['Body'].read()

    # Extract text from each page
    pdf_reader = PyPDF2.PdfReader(BytesIO(pdf_data))

    items_to_write = []

    for page_num, page in enumerate(pdf_reader.pages):
        page_text = page.extract_text()

        if not page_text.strip():
            print(f"Skipping empty page {page_num}")
            continue

        # Generate embedding
        embedding = get_embedding(page_text)

        # Prepare item for DynamoDB with user_id partition key
        item = {
            'user_id': user_id,  # Partition key for user isolation
            'document_page': f"{document_id}#{page_num}",  # Sort key: doc_id#page_num
            'document_id': document_id,
            'page_number': page_num,
            'page_text': page_text,
            'page_vector': convert_float_to_decimal(embedding),
            'document_type': 'pdf',
            's3_bucket': bucket,
            's3_key': key,
        }

        items_to_write.append(item)

    # Batch write to DynamoDB
    with table.batch_writer() as batch:
        for item in items_to_write:
            batch.put_item(Item=item)

    print(f"Indexed {len(items_to_write)} pages for document {document_id} (user: {user_id})")
    return len(items_to_write)

Important: Convert floats to Decimals – DynamoDB doesn't support native floats. This conversion is annoying but necessary.

Another very important detail — notice the batch_writer method? It’s another great feature of the Tableobject of the DynamoDB resource from boto3 — if the document has 10 pages, and we write the data after each page has been processed, it would take 10 database transactions to process the document.

By using batch_writer , we reduce it to ONE. 10x the improvement just by using one method.

Similarity Search: The Brute Force Approach

Here's where DynamoDB shows its limitations, where we're doing full linear scan with client-side similarity calculation to adjust to the DynamoDB table architecture and usage:

def cosine_similarity(vec1, vec2):
    """Calculate cosine similarity between two vectors using pure Python."""
    # Dot product
    dot_product = sum(a * b for a, b in zip(vec1, vec2))

    # Magnitude of vec1
    magnitude1 = math.sqrt(sum(a * a for a in vec1))

    # Magnitude of vec2
    magnitude2 = math.sqrt(sum(b * b for b in vec2))

    # Cosine similarity
    return dot_product / (magnitude1 * magnitude2)

def search_similar_pages(table_name: str, user_id: str, question_embedding: list, k: int = 5, document_ids: list = None):
    """
    Search for similar document pages using cosine similarity.
    ONLY searches within the specified user's data - complete isolation!

    Args:
        table_name: DynamoDB table name
        user_id: User ID for data isolation (partition key)
        question_embedding: Query vector
        k: Number of results to return
        document_ids: Optional list - search within specific documents only
    """
    table = dynamodb.Table(table_name)

    # Query only this user's data using partition key
    # This ensures User 1 can NEVER see User 2's data
    if document_ids and len(document_ids) > 0:
        # Search within specific documents for this user
        # Use FilterExpression with IN operator for multiple document IDs
        response = table.query(
            KeyConditionExpression='user_id = :uid',
            FilterExpression='document_id IN (' + ', '.join([f':doc{i}' for i in range(len(document_ids))]) + ')',
            ExpressionAttributeValues={
                ':uid': user_id,
                {f':doc{i}': doc_id for i, doc_id in enumerate(document_ids)}
            }
        )
    else:
        # Search all documents for this user
        response = table.query(
            KeyConditionExpression='user_id = :uid',
            ExpressionAttributeValues={':uid': user_id}
        )

    items = response['Items']

    # Handle pagination (DynamoDB returns max 1MB per query)
    while 'LastEvaluatedKey' in response:
        if document_ids and len(document_ids) > 0:
            response = table.query(
                KeyConditionExpression='user_id = :uid',
                FilterExpression='document_id IN (' + ', '.join([f':doc{i}' for i in range(len(document_ids))]) + ')',
                ExpressionAttributeValues={
                    ':uid': user_id,
                    {f':doc{i}': doc_id for i, doc_id in enumerate(document_ids)}
                },
                ExclusiveStartKey=response['LastEvaluatedKey']
            )
        else:
            response = table.query(
                KeyConditionExpression='user_id = :uid',
                ExpressionAttributeValues={':uid': user_id},
                ExclusiveStartKey=response['LastEvaluatedKey']
            )
        items.extend(response['Items'])

    print(f"Found {len(items)} pages for user {user_id}")

    # Calculate similarity for each item
    results = []
    for item in items:
        # Convert Decimal to float for numpy operations
        vector = [float(x) for x in item['page_vector']]
        similarity = cosine_similarity(question_embedding, vector)

        results.append({
            'page_number': int(item['page_number']),
            'page_text': item['page_text'],
            'score': float(similarity),
            'document_id': item['document_id'],
            'user_id': user_id  # Include for debugging (but never expose other users)
        })

    # Sort by similarity (highest first) and return top k
    results.sort(key=lambda x: x['score'], reverse=True)
    return results[:k]

Let’s be honest here — here is where time complexity starts playing a part. For 100 processed pages inside the table? Expect fast responses (500ms max), but if you have 10 000 pages? That’s where the user experience starts to take a hit because of the chosen architecture. When you hit this stage, it’s time to think about another database solution, most probably one of the vector databases.

But here's the thing: at small scale, this works fine. You're fetching data at DynamoDB speed (single-digit milliseconds), doing simple math in Python, and sorting. For a simple MVP, this is something we’ve expected and accepted as a trade-off.

This makes DynamoDB an excellent starting point, not a permanent solution — and that’s exactly the point.

The Complete Q&A Flow

Once we have relevant chunks from the DynamoDB table, we send them to an LLM to get an answer to the user’s question by doing the following steps:

build the context and prompt payloads, where we put the pages information received from the table and ask the LLM model to provide a clear answer to the user’s question
build a dynamic profile ID for the inference to access the Amazon Nova Micro model
receive, parse and return the response

def generate_answer(question: str, context_pages: list) -> str:
    """Generate answer using Bedrock with retrieved context."""
    # Build context from pages
    context = "\n\n".join([
        f"[Document: {page['document_id']}, Page {page['page_number'] + 1}]\n{page['page_text']}"
        for page in context_pages
    ])

    prompt = f"""You are a helpful assistant that answers questions based on provided document context.

Context from documents:
{context}

Question: {question}

Please provide a clear, accurate answer based only on the information in the context above. If the context doesn't contain enough information to answer the question, say so.

Answer:"""

    try:
        # Construct the inference profile ID based on region
        # For us-east-1 or us-west-2, use "us" prefix
        region = os.environ.get('BEDROCK_MODEL_REGION', 'us-east-1')
        region_prefix = 'us' if region in ['us-east-1', 'us-west-2'] else 'eu'
        inference_profile_id = f"{region_prefix}.amazon.nova-micro-v1:0"

        response = bedrock_runtime.invoke_model(
            modelId=inference_profile_id,
            body=json.dumps({
                "messages": [
                    {
                        "role": "user",
                        "content": [{"text": prompt}]
                    }
                ],
                "inferenceConfig": {
                    "max_new_tokens": 1024,
                    "temperature": 0.7,
                    "top_p": 0.9
                }
            })
        )

        # Return the response text
        response_body = json.loads(response['body'].read())
        return response_body['output']['message']['content'][0]['text']

    except Exception as e:
        print(f"Error generating answer: {str(e)}")
        raise

Cost Breakdown

One of the main reasons I chose DynamoDB for this RAG pipeline was cost predictability.

The estimates below assume on-demand capacity mode, which is ideal for MVPs and early-stage products with unpredictable traffic.

Assumptions

Average PDF size after processing: ~3 MB (worst case)
Average document split into 10 chunks/pages
Each chunk stored as a separate DynamoDB item (text + embedding + metadata)
Item size: ~3–4 KB per chunk
Region: us-east-1
No backups, streams, or global tables enabled

Monthly Cost Breakdown (Example)

Component	Usage Assumption	Monthly Cost
DynamoDB Storage	1,000 PDFs (~3 GB total)	~$0.75 (often free under 25 GB free tier)
DynamoDB Writes	10,000 items/month (ingestion)	~$0.04
DynamoDB Reads	10,000 similarity queries/month (brute-force scan)	~$0.08
S3 Storage	1,000 PDFs (3 GB)	~$0.07
Total (DynamoDB-related)	—	~$0.10–$1.00 / month

Even if you 10× the usage (10,000 documents or 100,000 queries/month), DynamoDB costs remain in the single-digit dollar range. At this stage, Bedrock embeddings and LLM inference will dominate your bill, not the database.

When This Approach Breaks (And How You’ll Know)

This DynamoDB-based RAG pipeline doesn’t fail silently — it gives you very clear warning signs when it’s time to start thinking about migrating to time to migrate to a purpose-built vector database. Here are a couple warning signs:

Pages per user: once a single user consistently exceeds ~1,000–2,000 stored chunks, brute-force similarity search starts to feel slow.
Query latency, especially p95; when it crosses ~3 seconds, users notice and trust drops.
Lambda memory usage — if your query Lambda needs to load thousands of vectors and regularly exceeds 50–60% memory just to sort results, you’re approaching a hard ceiling.
DynamoDB read capacity consumption — it will start climbing linearly with corpus size; not because DynamoDB is inefficient, but because you’re intentionally scanning more data per query.

None of these are bugs — they’re the natural limits of an O(n) retrieval strategy. The key advantage is that these limits are predictable, measurable, and give you plenty of time to migrate before the system becomes unusable.

Conclusion

Not every RAG system needs a vector database on day one.

If you’re building a POC project or a new feature you’re not yet sure users will care about, optimizing for lowest cost and fastest iteration matters more than performance. Using DynamoDB as a vector store lets you ship a fully functional, serverless RAG pipeline with almost no fixed infrastructure cost, predictable pricing, and minimal operational overhead.

The trade-offs are real and intentional as we’ve discussed. Similarity search is linear, latency grows with the size of the database. Eventually, this approach will hit a ceiling. But that ceiling is high enough for early-stage systems — and, more importantly, it’s visible and measurable. You’ll know when you’ve outgrown it long before it becomes a production problem.

The key takeaway is not that DynamoDB replaces vector databases. It doesn’t and it can’t. The main point of this blog post is to understand that you don’t have to pay for the vector database before you need it. Start simple, validate your idea and then migrate to new architecture which will support your new requirements.

If you’re building RAG for the first time, this approach will get you there faster and cheaper. And when it stops being enough, you’ll be in the best possible position to move on — with a working system, real users, and real constraints guiding the next step.

Eliminating Repetitive CRUD and Presigned URL Logic with AWS AppSync Pipeline Resolvers

Matia Rašetina — Fri, 02 Jan 2026 11:11:07 +0000

During the end of 2025, I wanted to end the year strong with learning and developing a project with a technology I wanted to learn for a long time, together with an AWS service which uses that same technology — GraphQL and AWS AppSync.

If you haven’t heard of GraphQL, it’s a data querying language developed by Meta, in which you can declare which data you want to fetch, update or create. It’s not tied to any specific database engine, like PostgreSQL or DynamoDB, so makes it even more interesting to use.

During 2025, I’ve built many projects with many operations and features, and many features were repetitive — CRUD operations in DynamoDB and pre-signed URL generation for object upload to S3 were the main ones. That’s why I decided to create a pattern with GraphQL to make CRUD operations and generation of pre-signed URLs easier and more straightforward.

Why GraphQL over REST? Here are the reasons:

overfetching — your REST API might provide additional information, which your frontend doesn’t need. By using GraphQL, you specifically define what data you need
underfetching — if you have multiple datasources to build one screen, GraphQL replaces them with only ONE request

This blog will not go into explaining GraphQL basics and will only talk about components of this pattern. If you are interested in learning GraphQL, please take a look at the official docs — https://graphql.org/learn/.

You can reuse this pattern in any project and you can be assured that your endpoints are working as expected right away. Only thing you need to do is to modify the object you are creating inside the DynamoDB table.

The code is available on this GitHub link: https://github.com/mate329/serverless-patterns/tree/mate329-pattern-appsync-lambda-s3-presigned-urls.

Architecture Overview & Understanding the Components

This pattern uses the following technologies and AWS Services:

Python and Python AWS CDK
AWS Lambda
AWS DynamoDB
AWS AppSync

Let’s understand the components of this pattern first.

The schema defines the contract that makes this pattern reusable: a single Note type with fields that can be progressively enriched by pipeline resolvers.

type Note {
  NoteId: ID!
  title: String
  content: String
  attachmentKey: String
  uploadUrl: String
  downloadUrl: String
}

type PaginatedNotes {
  notes: [Note!]!
  nextToken: String
}

type Query {
  allNotes(limit: Int, nextToken: String): PaginatedNotes!
  getNote(NoteId: ID!): Note
}

type Mutation {
  saveNote(NoteId: ID!, title: String!, content: String!, fileName: String): Note
  deleteNote(NoteId: ID!): Note
}

schema {
  query: Query
  mutation: Mutation
}

However, in this pattern, an additional step has been taken — I’ve implemented AppSync Pipeline Resolvers.

AppSync pipeline resolvers are a powerful but often overlooked feature of AWS AppSync. Think of them as a series of functions that execute in sequence for a single GraphQL operation, where each unique method can get the result of the previous step. Unlike traditional resolvers which interact with one datasource, pipeline resolvers let you orchestrate multiple AWS services in a single GraphQL query or mutation. The magic happens through the $ctx.prev.result variable in VTL (Velocity Template Language), which passes data from one function to the next, creating a clean data pipeline without writing orchestration code.

It removes so much boilerplate code — if you’re going to have multiple Lambdas for CRUD operations, you have to declare them inside the CDK code, give them permissions, set them up with correct environment variables and debug them if something happens. Then, of course, writing the Lambda code itself takes time to develop and to test.

By using AppSync pipeline resolvers, you let the pipeline handle everything for you. Less code = less worry that a bug might sneak in.

In this pattern, pipeline resolvers are used to:

Upload process — save the incoming data about the document inside the DynamoDB and call a Lambda to generate a pre-signed URL for document upload
Download process — fetch the data from the DynamoDB about the document and call a Lambda to generate a pre-signed URL for document download

VTL files will be shown in the following parts.

In addition, here is the base CDK code used for creating resources like AppSync, DynamoDB and S3 and creating data sources, so we can access the bucket and the table with AppSync:

from aws_cdk import (
    App,
    Stack,
    CfnOutput,
    Duration,
    RemovalPolicy,
    aws_dynamodb as ddb,
    aws_appsync as appsync,
    aws_lambda as lambda_,
    aws_s3 as s3,
)
from constructs import Construct

class NotesAppSyncStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # S3 bucket
        attachments_bucket = s3.Bucket(
            self, "NoteAttachmentsBucket",
            versioned=True,
            removal_policy=RemovalPolicy.DESTROY,
            auto_delete_objects=True,
            public_read_access=False,
            block_public_access=s3.BlockPublicAccess(
                block_public_acls=False,
                block_public_policy=False,
                ignore_public_acls=False,
                restrict_public_buckets=False
            ),
            cors=[
                s3.CorsRule(
                    allowed_headers=["*"],
                    allowed_methods=[
                        s3.HttpMethods.PUT,
                        s3.HttpMethods.POST,
                        s3.HttpMethods.DELETE,
                        s3.HttpMethods.GET
                    ],
                    allowed_origins=["*"],
                    max_age=3000
                )
            ]
        )

        # DynamoDB table
        table = ddb.Table(
            self,
            "DynamoDBNotesTable",
            partition_key=ddb.Attribute(name="NoteId", type=ddb.AttributeType.STRING),
            billing_mode=ddb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.RETAIN,
        )

        # Lambda function for S3 operations
        s3_lambda = lambda_.Function(
            self,
            "S3OperationsFunction",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="handler.handler",
            code=lambda_.Code.from_asset("PresignerLambda"),
            timeout=Duration.seconds(30),
            environment={
                "BUCKET_NAME": attachments_bucket.bucket_name,
                "TABLE_NAME": table.table_name,
            },
        )

        # Grant Lambda permissions
        attachments_bucket.grant_read_write(s3_lambda)
        table.grant_read_write_data(s3_lambda)

        # AppSync GraphQL API
        api = appsync.GraphqlApi(
            self,
            "Api",
            name="MyAppSyncApi",
            definition=appsync.Definition.from_file("graphql/schema.graphql"),
            authorization_config=appsync.AuthorizationConfig(
                default_authorization=appsync.AuthorizationMode(
                    authorization_type=appsync.AuthorizationType.API_KEY
                )
            ),
        )

        # API Key for Output
        api_key = appsync.CfnApiKey(self, "AppSyncApiKey", api_id=api.api_id)

        # Data Sources
        ddb_ds = appsync.DynamoDbDataSource(
            self,
            "AppSyncNotesTableDataSource",
            api=api,
            table=table,
            description="The Notes Table AppSync Data Source",
        )

        lambda_ds = appsync.LambdaDataSource(
            self,
            "S3OperationsDataSource",
            api=api,
            lambda_function=s3_lambda,
            description="Lambda for S3 presigned URL operations",
        )

The Upload Flow (saveNote mutation)

To understand this flow, we’ll go over it step-by-step.

Step 1: client calls the AppSync URL to generate the pre-signed URL for document upload

To save the note, AppSync calls the Presigner Lambda, which generates the pre-signed upload URL, here is the method used in the handler.py :

def generate_upload_url(event: Dict[str, Any]) -> Dict[str, Any]:
    """
    Generate a presigned URL for uploading a file to S3.

    Args:
        event: Contains 'noteId' and optional 'fileName'

    Returns:
        Dict containing 'uploadUrl' and 'attachmentKey'
    """
    note_id = event.get('noteId')
    file_name = event.get('fileName', 'attachment')

    if not note_id:
        raise ValueError("noteId is required")

    # Sanitize filename and create S3 key
    sanitized_filename = sanitize_filename(file_name)
    attachment_key = f"notes/{note_id}/{sanitized_filename}"

    try:
        # Generate presigned URL for PUT operation
        upload_url = s3_client.generate_presigned_url(
            'put_object',
            Params={
                'Bucket': BUCKET_NAME,
                'Key': attachment_key,
                'ContentType': get_content_type(sanitized_filename),
            },
            ExpiresIn=UPLOAD_URL_EXPIRATION,
            HttpMethod='PUT'
        )

        logger.info(f"Generated upload URL for note {note_id}, key: {attachment_key}")

        return {
            'uploadUrl': upload_url,
            'attachmentKey': attachment_key,
            'expiresIn': UPLOAD_URL_EXPIRATION
        }

    except ClientError as e:
        logger.error(f"Error generating upload URL: {str(e)}")
        raise Exception(f"Failed to generate upload URL: {str(e)}")

This request mapping template invokes the presigner Lambda and exposes its output (uploadUrl, attachmentKey) to subsequent pipeline functions via $ctx.prev.result.

The request VTL is:

{
  "version": "2018-05-29",
  "operation": "Invoke",
  "payload": {
    "operation": "generateUploadUrl",
    "noteId": $util.toJson($ctx.args.NoteId),
    "fileName": $util.toJson($util.defaultIfNull($ctx.args.fileName, "attachment"))
  }
}

Step 2: save the data to the DynamoDB table with the VTL payload

{
  "version": "2018-05-29",
  "operation": "PutItem",
  "key": {
    "NoteId": $util.dynamodb.toDynamoDBJson($ctx.args.NoteId)
  },
  "attributeValues": {
    "title": $util.dynamodb.toDynamoDBJson($ctx.args.title),
    "content": $util.dynamodb.toDynamoDBJson($ctx.args.content),
    "attachmentKey": $util.dynamodb.toDynamoDBJson($ctx.prev.result.attachmentKey),
    "uploadUrl": $util.dynamodb.toDynamoDBJson($ctx.prev.result.uploadUrl)
  }
}

Notice the $ctx.prev.result.uploadUrl ? That’s the payload which the Lambda has generated and provided to the second step of the resolver pipeline.

To build this pipeline inside the CDK code, here’s how you do it:

# Pipeline Function: Generate presigned URL for upload
generate_upload_url_fn = appsync.AppsyncFunction(
    self,
    "GenerateUploadUrlFunction",
    name="GenerateUploadUrlFunction",
    api=api,
    data_source=lambda_ds,
    request_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/generateUploadUrl.req.vtl"),
    response_mapping_template=appsync.MappingTemplate.lambda_result(),
)

# Pipeline Function: Save note to DynamoDB
save_note_to_ddb_fn = appsync.AppsyncFunction(
    self,
    "SaveNoteToDynamoDBFunction",
    name="SaveNoteToDynamoDBFunction",
    api=api,
    data_source=ddb_ds,
    request_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/saveNoteToDynamoDB.req.vtl"),
    response_mapping_template=appsync.MappingTemplate.dynamo_db_result_item(),
)

# Pipeline Resolver: saveNote (generate URL -> save to DDB)
# Here, we create the pipeline resolver which combines our pipeline parts,
# or rather pipeline functions
appsync.Resolver(
    self,
    "SaveNotePipelineResolver",
    api=api,
    type_name="Mutation",
    field_name="saveNote",
    pipeline_config=[generate_upload_url_fn, save_note_to_ddb_fn],
    request_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/saveNotePipeline.req.vtl"),
    response_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/saveNotePipeline.res.vtl"),
)

saveNotePipeline.req.vtl is an empty JSON {} , while the saveNotePipeline.res.vtl has the following content:

$util.toJson($ctx.prev.result)

The Download Flow (getNote query)

Step 1. Fetching the information about the note from the DynamoDB table

No VTL files for this step, as CDK supports fetching information directly inside the AppSync function definition!

Step 2: If the user has provided the attachment to the note, call the presigner Lambda to generate the download URL

The method to generate the download URL looks like:

def generate_download_url(event: Dict[str, Any]) -> Dict[str, Any]:
    """
    Generate a presigned URL for downloading a file from S3.

    Args:
        event: Contains 'attachmentKey'

    Returns:
        Dict containing 'downloadUrl' if attachment exists, empty dict otherwise
    """
    attachment_key = event.get('attachmentKey')

    # If no attachment key, return empty result (note has no attachment)
    if not attachment_key:
        logger.info("No attachment key provided, skipping download URL generation")
        return {}

    try:
        # Check if object exists
        try:
            s3_client.head_object(Bucket=BUCKET_NAME, Key=attachment_key)
        except ClientError as e:
            if e.response['Error']['Code'] == '404':
                logger.warning(f"Attachment not found: {attachment_key}")
                return {}
            raise

        # Generate presigned URL for GET operation
        download_url = s3_client.generate_presigned_url(
            'get_object',
            Params={
                'Bucket': BUCKET_NAME,
                'Key': attachment_key,
            },
            ExpiresIn=DOWNLOAD_URL_EXPIRATION,
            HttpMethod='GET'
        )

        logger.info(f"Generated download URL for key: {attachment_key}")

        return {
            'downloadUrl': download_url,
            'expiresIn': DOWNLOAD_URL_EXPIRATION
        }

    except ClientError as e:
        logger.error(f"Error generating download URL: {str(e)}")
        raise Exception(f"Failed to generate download URL: {str(e)}")

VTL request file, generateDownloadUrl.req.vtl , to fetch the download URL looks like:

## Only invoke Lambda if attachmentKey exists
#if($ctx.prev.result.attachmentKey)
{
  "version": "2018-05-29",
  "operation": "Invoke",
  "payload": {
    "operation": "generateDownloadUrl",
    "attachmentKey": $util.toJson($ctx.prev.result.attachmentKey)
  }
}
#else
## Pass through without invoking Lambda
{
  "version": "2018-05-29",
  "operation": "Invoke",
  "payload": {
    "operation": "skip"
  }
}
#end

while the response VTL file, generateDownloadUrl.res.vtl, looks like:

## Merge download URL into previous result if it exists
#if($ctx.result && $ctx.result.downloadUrl)
  #set($ctx.prev.result.downloadUrl = $ctx.result.downloadUrl)
#end
$util.toJson($ctx.prev.result)

CDK looks as following:

# Pipeline Function: Get note from DynamoDB
get_note_from_ddb_fn = appsync.AppsyncFunction(
    self,
    "GetNoteFromDynamoDBFunction",
    name="GetNoteFromDynamoDBFunction",
    api=api,
    data_source=ddb_ds,
    request_mapping_template=appsync.MappingTemplate.dynamo_db_get_item("NoteId", "NoteId"),
    response_mapping_template=appsync.MappingTemplate.dynamo_db_result_item(),
)

# Pipeline Function: Generate presigned URL for download (conditional)
generate_download_url_fn = appsync.AppsyncFunction(
    self,
    "GenerateDownloadUrlFunction",
    name="GenerateDownloadUrlFunction",
    api=api,
    data_source=lambda_ds,
    request_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/generateDownloadUrl.req.vtl"),
    response_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/generateDownloadUrl.res.vtl"),
)

# Pipeline Resolver: getNote (get from DDB -> generate download URL if attachment exists)
appsync.Resolver(
    self,
    "GetNotePipelineResolver",
    api=api,
    type_name="Query",
    field_name="getNote",
    pipeline_config=[get_note_from_ddb_fn, generate_download_url_fn],
    request_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/getNotePipeline.req.vtl"),
    response_mapping_template=appsync.MappingTemplate.from_file("resolvers/functions/getNotePipeline.res.vtl"),
)

Similarly to the saveNote VTL files, getNotePipeline.req.vtl is an empty JSON {} , while the getNotePipeline.res.vtl has the same content as saveNotePipeline.res.vtl .

Update and Delete Flows

Creating a delete flow is very easy, again, we can use already packaged method inside the AWS CDK. You can create the delete resolver via:

ddb_ds.create_resolver(
    "AppSyncDeleteNoteMutationResolver",
    type_name="Mutation",
    field_name="deleteNote",
    request_mapping_template=appsync.MappingTemplate.dynamo_db_delete_item("NoteId", "NoteId"),
    response_mapping_template=appsync.MappingTemplate.dynamo_db_result_item(),
)

, while the update flow is built on the saveNote mutation — you just provide the noteId field and update the fields. Examples of all methods in this pattern are shown in the following chapter.

Testing It Out

Let’s go over the whole pattern and show you examples of how this works.

Create Note

We will start with the saveNote mutation. Here is the GraphQL mutation:

mutation SaveNote {
    saveNote(
        NoteId: "123"
        title: "My First Note"
        content: "This is the content of my note"
        fileName: "test.txt"
    ) {
        NoteId
        title
        content
        attachmentKey
        uploadUrl
        downloadUrl
    }
}

and here is the expected response:

and when you create a sample test.txt file and use the provided upload URL, you will get the following interface:

Get Note

With the following GraphQL payload, you will get the newly created note:

query GetNote {
    getNote(NoteId: "123") {
        NoteId
        title
        content
        attachmentKey
        downloadUrl
    }
}

and it should look like this in Postman, or any other API testing tool:

In case you want to fetch all notes, the following GraphQL query can be used. The result will be similar as the previous one, so an image won’t be provided. Here is the query though:

query GetAllNotes {
    allNotes(limit: 10, nextToken: null) {
        notes {
            NoteId
            title
            content
            attachmentKey
            uploadUrl
            downloadUrl
        }
        nextToken
    }
}

Update Note

As mentioned, Update note flow uses the saveNote mutation

mutation UpdateNote {
  saveNote(
    NoteId: "123"
    title: "Updated Title"
    content: "Updated content"
    fileName: "test.txt"
  ) {
    NoteId
    title
    content
    attachmentKey
  }
}

here is the execution example:

Delete Note

Lastly, here is the usage of the deleteNote mutation:

mutation DeleteNote {
    deleteNote(NoteId: "123") {
        NoteId
    }
}

, together with the execution results:

Security and Cost

API Key vs AWS Cognito — Which one to choose?

Regarding security, in this pattern the API key is used, for the sake of simplicity. AWS Cognito can be used to protect the GraphQL endpoint, but I didn’t want to make the pattern dependent on another AWS service.

However, let’s compare the differences in between choosing the API key and Cognito as the security choice.

By using the API key, you have the following pros:

very easy to implement (just a few lines of code inside the CDK to fetch the key for your GraphQL endpoint)
great for quick prototyping or testing
no additional AWS services required and no additional AWS services cost

There are cons to this approach too:

no user identity - you cannot tell who made the request
not suitable for real-world production application, as you need to implement additional logic to filter out requester’s data
no granular permissions - everyone has access to everything AppSync does

On the other hand, here are the pros of using Cognito as the authorizer:

individual user identities - you know exactly who's making each request
users can only access their data
fine-grained access control using IAM policies or Lambda authorizers, making your system more secure

Cons of this approach are:

possible additional cost in production - Cognito is free until you have more than 50000 users
more complex to setup, both for frontend and backend developers
another AWS service dependency

The choice is up to you!

Cost — Will it break the bank?

When talking about cost, AppSync is in the Free Tier and you get the following number of operations:

250,000 query or data modification operations
250,000 real-time updates
600,000 connection-minutes

This makes it a very good choice, especially if you expect traffic lower than this.

Of course, if you use AWS Lambda REST API, it most probably will be cheaper, but by using this pattern you get off the ground quicker than developing Lambdas for every CRUD operation + you get the benefits I’ve talked about at the beginning of the blog.

Conclusion

This pattern shows how GraphQL with the combination of AWS CDK and AWS AppSync pipeline resolvers can be used to move beyond simple CRUD APIs and into orchestrated backend workflows.

By combining AppSync, DynamoDB, Lambda and S3, this pattern models one of the most common application flows as a single GraphQL operation, while still keeping the implementation modular and reusable.

As already mentioned, the main advantage of this approach is not just fewer API calls, but reduced backend boilerplate. Instead of writing and maintaining multiple Lambdas for repetitive CRUD logic and presigned URL handling, AppSync pipeline resolvers handle everything for you.

With all of these positives, there are some drawbacks as well — this pattern is not a replacement for REST, as it introduces it’s own complexity, especially around VTL files and it makes most sense when you already rely on AWS services. For small workflows, a traditional Lambda may still be the simpler choice.

If you find yourself repeatedly building the same serverless backend features — CRUD operations, file uploads, and basic orchestration — this pattern can serve as a solid foundation. It is intentionally minimal, easy to adapt, and designed to be extended as application requirements grow.

The Complete Guide to AWS Lambda Aliases, Versions, and Canary Deployments (With CDK Examples)

Matia Rašetina — Fri, 28 Nov 2025 09:00:00 +0000

Deploying a new Lambda code to your AWS environment shouldn’t be stressful at all, but for some teams, it is. If you ever pushed a quick fix to production and straight away went to CloudWatch logs to see if the Lambda is failing, you know the feeling. When updating the Lambda code, it happens instantly. If there is a bug in the new code, every user feels it straight away.

That’s why tools like Lambda Versions, Aliases and Canary Deployments come in. When used correctly, they give you a way of rolling out new code changes gradually, observing the impact and automatically starting the roll back process if something happens.

No downtime. No fire drills. No late-night debugging sessions.

In this guide, you'll learn:

What Lambda versions and aliases actually do
How traffic routing works
How CodeDeploy turns aliases into safe, automated canary releases
How to implement everything with AWS CDK
The pitfalls and best practices you should know

This blog post is a continuation of a previous post about CRUD Lambdas with PynamoDB. Here, we are going to enable versioning, alias and deployment for CreatePropertyLambda , however the full code for all CRUD Lambdas is available on this link.

Why Safe Lambda Deployments Matter

Most of the Lambda deployments follow the most basic flow:

a developer deploys the code via CDK, SAM or any other tool
AWS takes the new snapshot of the incoming code
AWS switches all traffic to the new deployed Lambda

It’s fast, however it makes it very dangerous for the stability of the application, as you don’t know if there will be any errors which went unnoticed during testing.

With aliases and canary routing, you shift from “everything at once” to:

Send 10% of traffic to the new version
Watch for errors
Promote or roll back automatically

This gives you:

Zero downtime deployments
Automatic recovery when something breaks
Gradual exposure to real production traffic
Safety without slowing delivery

Let’s take a deeper dive into the basics and understand them before continuing to the CDK code.

Lambda Versions, Aliases and Routing — The Essentials

Before we begin, let’s go over some base points, which are need-to-know.

Versions

A version is an immutable snapshot of that code, meaning once deployed, it will forever have that version number and it can only be updated by another version.

One Lambda version contains multiple parts which are vital for Lambda runtime:

source code
layers and dependencies
environment variables
runtime settings, like Lambda timeout

This immutability is why versions are the backbone of safe deployments.

Aliases

An alias is a named pointer — like prod, beta, or v1 — that points to one or more Lambda versions.

It is very important to understand — API Gateway, Step Functions, EventBridge rules or any other AWS service should call your alias, not the version directly, because aliases enable us to decouple deployments from references, allow traffic shifting and rollback and aliases prevent needing to update API Gateway integrations.

Weighted Routing + Why use Canary Deployments?

AWS lets you attach a routing configuration to an alias — for example, we deploy a new Lambda version and we can route the traffic to both old and new versions:

90% of traffic → version 1
10% of traffic → version 2

This simple mechanism is the foundation of canary deployments.

We use canary deployments to route a fraction of the production traffic to the newly deployed resource. If for any reason the new version of the resource fails, only a small slice of users is impacted and the system will recognize the errors and re-route the experimental users back to the fully working code.

There are other benefits to using canary deployments:

Zero downtime
Safe experimentation with new logic or dependencies
Easy rollback by flipping the alias back

For more information how the canary deployment works, please check out my blog post about it, you can see it by clicking the link here.

How Lambda Aliases Enable Canary Releases

The full canary release process is as following:

new Lambda code gets deployed → AWS automatically creates a new version, let’s call it version 2
Lambda alias prod is pointing to version 1
prod alias gets updated to route the traffic:
- 90% of the traffic → version 1
- 10% of the traffic → version 2
CodeDeploy and CloudWatch are working together to monitor the incoming traffic to version 2
- if anything breaks and errors spike, CloudWatch sends the notification to CodeDeploy to stop the deployment process and to reroute all traffic to the older version (in this case, version 1)
- if everything looks good, shift the traffic from version 1 to version 2 and version 2 becomes a de facto main carrier of prod alias traffic

With this approach, users experience no downtime and you can be assured that in case of failure, your system is still online!

Setting Up Canary Deployments with AWS CDK

Step 1: Enable Lambda versioning

You want Lambda to publish a new version only when the code changes*.*

A common trick: embed a hash into the function description.

# Method used to get the hash of the Lambda, so we confirm if
# the code changed at all.
# No need to start a canary deployment if the code didn't change
def _get_code_hash(self, directory: str) -> str:
        """Generate a hash of all files in the Lambda code directory."""
        hash_md5 = hashlib.md5()
        for root, dirs, files in sorted(os.walk(directory)):
            for file in sorted(files):
                if file.endswith(('.py', '.json', '.txt', '.yaml', '.yml')):
                    file_path = os.path.join(root, file)
                    with open(file_path, 'rb') as f:
                        hash_md5.update(f.read())
        return hash_md5.hexdigest()[:8]

# Lambda functions with code hash in description for versioning
create_property_code_dir = "./CreatePropertyHandler"
create_code_hash = self._get_code_hash(create_property_code_dir)

create_property_fn = _lambda.Function(
    self, "CreatePropertyLambda",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="handler.handler",
    code=_lambda.Code.from_asset(create_property_code_dir),
    layers=[layer],
    description=f"v-{create_code_hash}",  # Hash in description triggers new version only when code changes
    environment={
        "PROPERTY_TABLE_NAME": self.property_table.table_name,
        "PROPERTY_BUCKET": self.property_bucket.bucket_name,
        "ALLOWED_ORIGIN": "*"
    },
    timeout=Duration.seconds(30)
)

create_version = create_property_fn.current_version

Step 2 — Create a Lambda Alias

create_alias = _lambda.Alias(
    self, "CreatePropertyProdAlias",
    alias_name="prod",
    version=create_version,
)

This is the entry point of accessing the CreateProperty Lambdas which will be deployed.

Step 3 — Creating a CloudWatch Alarm

This alarms will trigger a rollback if the function is reporting any errors:

create_error_alarm = cloudwatch.Alarm(
    self, "CreatePropertyErrorAlarm",
    metric=create_property_fn.metric_errors(
        period=Duration.seconds(30),
        statistic="Sum"
    ),
    threshold=1,
    evaluation_periods=2,
    datapoints_to_alarm=2,
    alarm_name="create-property-canary-errors"
)

A very useful feature in CloudWatch is that you can monitor not just Lambda errors, but also:

latency
throttles
create metrics of your own - you can read more about custom metrics by clicking on the link here

Step 4 — Adding a Canary Deployment process with CodeDeploy

codedeploy.LambdaDeploymentGroup(
    self,
    "CreatePropertyCanaryDeployment",
    alias=create_alias,
    deployment_config=codedeploy.LambdaDeploymentConfig.CANARY_10_PERCENT_5_MINUTES,
    alarms=[create_error_alarm],
    auto_rollback=codedeploy.AutoRollbackConfig(
        failed_deployment=True,
        stopped_deployment=True,
    ),
)

The configuration I’ve used is:

route 10% of traffic to the new version for 5 minutes
- if no errors, shift to 100% instantly
- if errors, roll back to the previous version

When you start your deployment via cdk deploy , if you go to your AWS Console and go to CodeDeploy, you will see the following interface and see how your deployment is going:

I’ve written about the deployment configuration in my other blog post which focuses more on Canary Deployments. If you are interested in more details, please check out my other post by clicking on the link here.

Best Practices and Lessons Learned

A couple of lessons and best practices I’ve learned:

use aliases for everything in production
- never point to a specific Lambda version, only aliases
canary deployments don’t work for all triggers
- SQS, Kinesis and DynamoDB streams are not supported — canaries can only apply where AWS can split traffic, like API Gateway and Application Load Balancer
versioning increases storage
- old Lambda code versions can accumulate, so consider lifecycle cleanup of unused Lambda versions
never forget about cold starts
- it’s very useful to have a latency alarm during deployment, however, keep in mind that Lambda’s cold starts + regular execution time can trigger the alarm too!

Conclusion

Your Lambda deployments don’t have to be risky. By combining all of the points we’ve went over in this blog post, like versions, aliases and canary deployments, you can get a predictable, reversible and fully automated rollouts — with zero downtime!

In a production-grade system, implementation of this pattern should be a no-brainer.

PynamoDB Tutorial: Build Production-Ready DynamoDB CRUD APIs in AWS Lambda

Matia Rašetina — Tue, 25 Nov 2025 09:00:00 +0000

I've been building on AWS for multiple years now, and one of the most repetitive parts I found was writing crude operations for DynamoDB. Every project starts the same:

use boto3 to define the table
define a big JSON payload JUST to fetch some data from the table
copy-and-paste methods from previous projects for all CRUD operations, basically adding boilerplate code

That's when I found out about a library called PynamoDB , a lightweight ORM for DynamoDB tables and that changed the way I manipulate table’s data.

Welcome to the part two of creating code for a real estate platform where in the first post we went over a pipeline of labeling main parts of the properties and now we're gonna go over Lambdas which are responsible for CRUD operations.

The code in this post will be a shortened version than the one from the full project, just to make it as simple as possible. The full code used on the platform will be available in the next blog posts.

In this post, I’ll break down how I put together a simple CRUD stack using:

PynamoDB for the data model
Lambda functions for the business logic and interaction with the DynamoDB table via PynamoDB model

The whole goal is ease of use — a DynamoDB-backed CRUD pattern that feels as natural as working with a traditional ORM, but with all the serverless benefits of AWS.

AWS CDK code won’t be mentioned in this blog post, however it will be available on the GitHub repository which you can find by clicking on this link.

Which AWS resource are we using?

All resources will be created via AWS CDK in Python.

We will start off with AWS Lambda - here is a short description of the Lambdas and what they do:

Create Property Lambda - enter received property information, put it into the DynamoDB table and provide the user pre-signed URLs for property image uploads to the S3 bucket
Get Property Lambda - read all or a specific property from the database
Update Property Lambda - update the specific property entry which is inside the database
Delete Property Lambda - delete all images and entry inside the database

Of course, since we are showcasing the PynamoDB library, we need to use DynamoDB as our database.

Why should I use PynamoDB instead of boto3.client?

Well, great question! Take a look at the Python snippets which compares the syntax for both libraries:

With boto3.client:

item = {
    'PK': {'S': f'PROPERTY#{property_id}'},
    'price': {'N': '500000'},
    'title': {'S': 'Beautiful Home'},
    'createdAt': {'S': '2024-01-15T10:30:00Z'}
}
dynamodb.put_item(TableName='Properties', Item=item)

With PynamoDB:

property_item = PropertyModel(
    property_id=property_id,
    price=500000,
    title='Beautiful Home',
    createdAt=datetime.now()
)
property_item.save()

Notice how PynamoDB handles type conversion automatically - no more wrapping everything in {'S': ...} or {'N': ...} dictionaries!

Designing the PynamoDB model

Here is the PynamoDB model which I’ve used for my real estate platform:

from datetime import datetime, timezone
from os import environ
from pynamodb.models import Model
from pynamodb.attributes import (
    UnicodeAttribute, NumberAttribute, UTCDateTimeAttribute, MapAttribute,
    ListAttribute, BooleanAttribute, JSONAttribute
)

class PropertyModel(Model):
    class Meta:
        table_name = environ.get('PROPERTY_TABLE_NAME', 'SimplifiedPropertyTable')
        region = environ.get('AWS_REGION', 'eu-central-1')

    PK = UnicodeAttribute(hash_key=True)
    SK = UnicodeAttribute(range_key=True)

    id = UnicodeAttribute()
    title = UnicodeAttribute(null=True)
    description = UnicodeAttribute(null=True)
    price = NumberAttribute(null=True)
    address = UnicodeAttribute(null=True)
    city = UnicodeAttribute(null=True)
    state = UnicodeAttribute(null=True)
    zip = UnicodeAttribute(null=True)

    createdAt = UTCDateTimeAttribute(null=True)
    updatedAt = UTCDateTimeAttribute(null=True)

    def __init__(self, property_id: str = None, **kwargs):
        now = datetime.now(timezone.utc)
        if property_id:
            kwargs.setdefault('PK', f"PROPERTY#{property_id}")
            kwargs.setdefault('SK', "METADATA")
            kwargs.setdefault('id', property_id)
            kwargs.setdefault('createdAt', now)
            kwargs.setdefault('updatedAt', now)
        super().__init__(**kwargs)

You can see that there are many information here, from numbers, booleans, lists, strings.

When using PynamoDB, it’s important to understand how PynamoDB interprets your data. You can see many new objects like UnicodeAttribute , BooleanAttribute , ListAttribute and others. Here are the Python-PynamoDB equivalents:

NumberAttribute - integer or float
ListAttribute - list or array
UnicodeAttribute - string
BooleanAttribute - boolean
JSONAttribute - a JSON object
MapAttribute - Python’s dictionary
UTCDateTimeAttribute - Python’s datetime.now() method

In this case, every single information about the property was put inside only one database. You may ask “Why?”

Two reasons which were crucial for this project:

makes the usage of DynamoDB much cheaper — if we had 2 or more tables, to fetch the data we’d need to call multiple tables, but this way we only call one, which reduces cost
no schema migrations - any new data can be easily added to the already existing DynamoDB entry. As I wanted to complete this project as soon as possible, I kept adding data to the table. DynamoDB enabled me that without complaining once.

Also, do you notice the Meta class?

The Meta class is incredibly powerful because it externalizes your configuration. Here's what's happening:

class Meta:
    table_name = environ.get('PROPERTY_TABLE_NAME', 'SimplifiedPropertyTable')
    region = environ.get('AWS_REGION', 'eu-central-1')

This means you can use the same code across environments:

Development:

export PROPERTY_TABLE_NAME=dev-properties-table
export AWS_REGION=us-east-1

Production:

export PROPERTY_TABLE_NAME=prod-properties-table
export AWS_REGION=eu-central-1

No code changes needed - just different environment variables. This is especially useful when deploying with CDK, as you can set these per stack/stage automatically.

CRUD operations with the PynamoDB model

Create DynamoDB Entry Lambda

Here is a very simple method which leverages the created PynamoDB model and saves the property information with ease, while creating pre-signed URLs for property images - just use the .save() method:

def handler(event, context):
    """
    Create a new property and return presigned URLs for image uploads
    """
    try:
        # Parse request body
        body = json.loads(event.get('body', '{}'))

        # Validate required fields
        required_fields = ['title', 'description', 'price', 'address', 'city', 'state', 'zip']
        missing_fields = [field for field in required_fields if field not in body]

        if missing_fields:
            logger.warning("Missing required fields", extra={"missing": missing_fields})
            return {
                "statusCode": 400,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": f"Missing required fields: {', '.join(missing_fields)}"})
            }

        # Generate unique property ID
        property_id = str(uuid.uuid4())

        # Create property in DynamoDB
        property_item = PropertyModel(
            property_id=property_id,
            title=body['title'],
            description=body['description'],
            price=body['price'],
            address=body['address'],
            city=body['city'],
            state=body['state'],
            zip=body['zip']
        )
        property_item.save()

        logger.info("Property created", extra={"property_id": property_id})

        return {
            "statusCode": 201,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({
                "property_id": property_id,
                "property": property_item.to_dict()
            })
        }

    except Exception as e:
        logger.exception("Error creating property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to create property"})
        }

Get DynamoDB Entry Lambda

There are 2 flows for this Lambda:

if the property ID is provided in the request, query the database for information just for that property
scan the table and get all the data

To scan the database, use the pynamodb_model.scan() method or to fetch the precise item from the database, use the pynamodb_model.get('ID_of_item_you_want_to_fetch'):

def handler(event, context):
    """
    Get a single property by ID or list all properties
    """
    try:
        # Check if property_id is provided in query parameters
        query_params = event.get('queryStringParameters') or {}
        property_id = query_params.get('propertyId')

        if property_id:
            # Get single property and handle the case
            # if the property requested doesn't exist
            try:
                property_item = PropertyModel.get(
                    hash_key=f"PROPERTY#{property_id}",
                    range_key="METADATA"
                )
            except PropertyModel.DoesNotExist:
                property_item = None

            if not property_item:
                logger.warning("Property not found", extra={"property_id": property_id})
                return {
                    "statusCode": 404,
                    "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                    "body": json.dumps({"error": "Property not found"})
                }

            logger.info("Property retrieved", extra={"property_id": property_id})
            return {
                "statusCode": 200,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"property": property_item.to_dict()})
            }

        else:
            # List all properties (scan operation)
            properties = []
            for item in PropertyModel.scan():
                properties.append(item.to_dict())

            logger.info("Properties listed", extra={"count": len(properties)})
            return {
                "statusCode": 200,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({
                    "properties": properties,
                    "total": len(properties)
                })
            }

    except Exception as e:
        logger.exception("Error retrieving property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to retrieve property"})
        }

Important note on scanning: The PropertyModel.scan() operation shown above works great for small datasets, but scanning becomes expensive and slow as your table grows. In production, consider:

Pagination: For large result sets, use last_evaluated_key to paginate:

results = PropertyModel.scan(limit=20)
for item in results:
    properties.append(item.to_dict())

# Get the last evaluated key for next page
if results.last_evaluated_key:
    next_results = PropertyModel.scan(
        limit=20,
        last_evaluated_key=results.last_evaluated_key
    )

Query instead of Scan: If you frequently filter by city or state, add a Global Secondary Index (GSI) and use PropertyModel.city_index.query('Vienna') instead of scanning everything.
Limit for safety: Always set a limit in production to prevent timeout issues: PropertyModel.scan(limit=100)

For my real estate platform with hundreds of properties, I added pagination and GSIs for common queries - I'll cover this in a future post!

Update DynamoDB Entry Lambda

Here is the code for updating the property already entered in the platform by using pynamodb_model.update() method:

def handler(event, context):
    """
    Update an existing property
    """
    try:
        # Get property ID from path parameters
        property_id = event.get('pathParameters', {}).get('id')

        if not property_id:
            return {
                "statusCode": 400,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property ID is required"})
            }

        # Parse request body
        body = json.loads(event.get('body', '{}'))

        # Get existing property
        try:
            property_item = PropertyModel.get(
                hash_key=f"PROPERTY#{property_id}",
                range_key="METADATA"
            )
        except PropertyModel.DoesNotExist:
            property_item = None

        if not property_item:
            logger.warning("Property not found", extra={"property_id": property_id})
            return {
                "statusCode": 404,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property not found"})
            }

        # Update fields
        update_actions = []
        updatable_fields = ['title', 'description', 'price', 'address', 'city', 'state', 'zip']

        for field in updatable_fields:
            if field in body:
                update_actions.append(
                    getattr(PropertyModel, field).set(body[field])
                )

        # Always update the updatedAt timestamp
        update_actions.append(
            PropertyModel.updatedAt.set(datetime.now())
        )

        if update_actions:
            property_item.update(actions=update_actions)

        # Refresh the item to get updated values
        property_item.refresh()

        logger.info("Property updated", extra={"property_id": property_id})
        return {
            "statusCode": 200,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"property": property_item.to_dict()})
        }

    except Exception as e:
        logger.exception("Error updating property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to update property"})
        }

Delete DynamoDB Entry Lambda

The request should provide the property ID, and delete the property details by using the pynamodb_model.delete() method:

def handler(event, context):
    """
    Delete a property by ID
    """
    try:
        # Get property ID from path parameters
        property_id = event.get('pathParameters', {}).get('id')

        if not property_id:
            return {
                "statusCode": 400,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property ID is required"})
            }

        # Get existing property
        # and catch the PropertyModel.DoesNotExist error 
        # if the property ID is not found inside the DynamoDB
        try:
            property_item = PropertyModel.get(
                hash_key=f"PROPERTY#{property_id}",
                range_key="METADATA"
            )
        except PropertyModel.DoesNotExist:
            property_item = None

        if not property_item:
            logger.warning("Property not found", extra={"property_id": property_id})
            return {
                "statusCode": 404,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property not found"})
            }

        # Delete the property
        property_item.delete()

        logger.info("Property deleted", extra={"property_id": property_id})
        return {
            "statusCode": 200,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"message": "Property deleted successfully"})
        }

    except Exception as e:
        logger.exception("Error deleting property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to delete property"})
        }

Error and Race Condition Handling

One thing to be aware of when building production CRUD operations: concurrent updates.

Imagine two users trying to update the same property price simultaneously. Without proper handling, the last write wins and one update gets lost.

PynamoDB supports conditional writes using conditions:

from pynamodb.expressions.condition import Condition

def handler(event, context):
    # ... existing code to get property_item ...

    # Only update if the price hasn't changed since we read it
    original_price = property_item.price

    try:
        property_item.update(
            actions=[PropertyModel.price.set(new_price)],
            condition=(PropertyModel.price == original_price)
        )
    except PropertyModel.UpdateError:
        return {
            "statusCode": 409,
            "body": json.dumps({
                "error": "Property was modified by another user. Please refresh and try again."
            })
        }

This is especially important for sensitive fields like price or availability status. For my platform, I use conditional updates when changing property status to prevent double-booking scenarios.

Deployment process

Very easy! You can take the GitHub repository provided and just run cdk deploy inside your terminal window!

Just make sure your AWS credentials are correctly setup on your machine.

Future work & Conclusion

Now that we went over how to use PynamoDB ORM usage, the next questions should be:

how do we want to protect our CreateProperty , UpdateProperty and DeleteProperty Lambdas?
- my choice? AWS Cognito makes it very easy to protect the endpoints via an AWS API Gateway Authorizer method
single Lambda handling multiple HTTP methods?
- everything is in one place, however code maintenance might become an issue
Improving Lambda performance by implementing Global Secondary Indexes inside the DynamoDB table
- this could be implemented based on which data the users mostly get

Ever since I’ve found out about PynamoDB, I’ve been using it constantly in every project. It just makes it so easy to add, access and modify the data which is in the DynamoDB table. Hopefully it helps you in your coding!

Thank you for reading! See you in the next post!

Designing a Scalable Image Labeling Workflow with AWS S3, SQS, Lambda, Rekognition and Bedrock

Matia Rašetina — Mon, 17 Nov 2025 09:18:42 +0000

Introduction

Managing the data on a real estate platform at scale can be a very challenging task — from tracking properties and images to generating useful summaries for buyers and sellers. Traditional real estate applications use very simple architecture, without adapting to the newest AI available features.

That’s why I’ve decided to create a new real estate platform. The one where:

you will learn how to batch data and lower the Lambda bill by up to 10x
integration of AWS Rekognition and Bedrock
learn how to create real-world architecture pattern that’s not beginner-level

This new platform will be full with AI features, powered by technologies like NextJS for the frontend and AWS for the backend. We’ll take this project apart, take a deep dive into all parts of this project, from CI/CD deployments, mock testing to deploying via Lambda aliases and canary deployments.

In this post, I’ll walk through a serverless, event-driven backend I built entirely with AWS CDK in Python, designed to create labels to describe the images of properties and summaries of properties based on those labels. The automatic image analysis is done by using Amazon Rekognition, and AI-generated property summaries with AWS Bedrock. The system leverages S3, Lambda, SQS and DynamoDB to create a scalable, cost-efficient, and resilient solution that can process thousands of images and generate real-time insights — all without managing servers.

By the end of this post, you’ll understand how to design modular serverless pipelines, implement event-driven workflows, and integrate AI-powered summarization into your own applications.

All code can be found at the GitHub repository by clicking on the this link.

Let’s start!

Architecture Overview of AWS Rekognition Image Labeling

Here is an architecture diagram, which executes when the user uploads an image of the property to the pre-signed URL provided by our property service for creating an property inside our system.

When the image is fully uploaded to S3, S3 sends an Event Notification to SQS with the information about the uploaded image.

Now, imagine receiving 100 images and not having them batched. That would mean that the system needs to spin up 100 Lambdas, but if we configure the batching setting to 10 that would mean 10 times less than the invocations which lowers cost.

That’s why the backend is configured to batch up to 10 Event Notifications inside the queue, and those 10 messages are sent to the land to lower cold starts and to lower cost. It’s worth mentioning that this setting can be changed on your project’s requirements.

The Lambda takes each of the image and send it to AWS Rekognition for it to do the image analysis, to create labels based on the image. Also, in the payload we ask Rekognition to only return labels in which it’s 75% confident. After the labels are received, they are saved inside the DynamoDB database for that particular image

Code Deep-Dive - AWS Rekognition Label Detection

Let's take a closer look into the configuration of this pipeline and how can you do it yourself. This section will only contain snippets, but you can take a closer look into the app.py file which can be found on GitHub.

Here is the CDK code for initializing the Lambda resource, the SQS queue, the DynamoDB table and the S3 bucket:

# S3 bucket for property images
self.property_bucket = s3.Bucket(
    self, "PropertyBucket",
    versioned=True,
    removal_policy=RemovalPolicy.DESTROY,
    auto_delete_objects=True,
    public_read_access=False,
    block_public_access=s3.BlockPublicAccess(
        block_public_acls=False,
        block_public_policy=False,
        ignore_public_acls=False,
        restrict_public_buckets=False
    ),
    cors=[
        s3.CorsRule(
            allowed_headers=["*"],
            allowed_methods=[
                s3.HttpMethods.PUT,
                s3.HttpMethods.POST,
                s3.HttpMethods.DELETE,
                s3.HttpMethods.GET
            ],
            allowed_origins=["*"],
            max_age=3000
        )
    ]
)

# Property Images Table - stores individual image metadata and analysis
self.property_images_table = dynamodb.Table(
    self, "PropertyImagesTable",
    partition_key=dynamodb.Attribute(name="PK", type=dynamodb.AttributeType.STRING),
    sort_key=dynamodb.Attribute(name="SK", type=dynamodb.AttributeType.STRING),
    billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
    removal_policy=RemovalPolicy.DESTROY,
)

# SQS Queue for image upload events - batches multiple uploads
image_queue = sqs.Queue(
    self, "ImageAnalysisQueue",
    visibility_timeout=Duration.seconds(360),
    retention_period=Duration.days(14),
    removal_policy=RemovalPolicy.DESTROY
)

# Image Analysis Lambda - Triggered by SQS with batching
image_analysis_code_dir = "./ImageAnalysisHandler"

image_analysis_fn = _lambda.Function(
    self, "ImageAnalysisLambda",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="handler.handler",
    code=_lambda.Code.from_asset(image_analysis_code_dir),
    environment={
        "PROPERTY_IMAGES_TABLE": self.property_images_table.table_name,
        "PROPERTY_BUCKET": self.property_bucket.bucket_name,
        "ALLOWED_ORIGIN": "*",
        "CODE_VERSION": image_code_hash
    },
    timeout=Duration.seconds(60)
)

Also, we need to add the necessary IAM permissions to the Lambda to have read-only access to images inside S3, write-only permissions to the PropertyImagesTable , and to be able to call the AWS Rekognition API:

# Enabling read-only access to S3
self.property_bucket.grant_read(image_analysis_fn)

# Enabling write-only access to the PropertyImages Table
self.property_images_table.grant_write_data(image_analysis_fn)

# Enabling the Lambda to be able to call the Rekognition 
# service to detect labels
image_analysis_fn.add_to_role_policy(
            iam.PolicyStatement(
                actions=["rekognition:DetectLabels"],
                resources=["*"]
            )
        )

Now, we need to tell our CDK code to:

create an Event Notification on image upload to S3
send that Event Notification for the particular image to SQS
tell SQS queue to invoke our ImageAnalysisHandler Lambda with 10 batched notifications at most

We can do that with the following code:

# S3 Event Notification
self.property_bucket.add_event_notification(
    s3.EventType.OBJECT_CREATED, # Create the notification only when an object is uploaded
    s3n.SqsDestination(image_queue), # Put the notification inside the SQS queue
    s3.NotificationKeyFilter(prefix="properties/") # Only send the Notificaiton if it's uploaded to the "properties" folder
)

# Lambda SQS Event Source
image_analysis_fn.add_event_source(
    lambda_event_sources.SqsEventSource(
        image_queue, # Take payloads from the SQS queue
        batch_size=10, # Batch up to 10 payloads to reduce cost
        max_batching_window=Duration.seconds(2), # Wait for 2 seconds to fill up the batch
        report_batch_item_failures=True,
    )
)

Now that we’ve created our infrastructure, let’s take a look at the main method inside our Lambda which takes the images from S3 and gets the labels from Rekognition. As previously said, if you need the full Lambda code, it is available on the provided GitHub link.

What makes using Rekognition great is that you can tell it the image location inside the created S3 bucket and it’s going to do everything else for you - no extra payload fields!

I’ve configured my payload to make Rekognition return 5 labels at most and it needs to be at least 75% confident in it’s label recognition.

def analyze_image_with_rekognition(bucket_name: str, object_key: str) -> dict:
    """
    Use AWS Rekognition to detect labels only

    Returns:
        dict: Analysis results with labels
    """
    analysis = {
        "labels": []
    }

    try:
        # Detect labels
        logger.info("Detecting labels")
        labels_response = rekognition_client.detect_labels(
            Image={
                'S3Object': {
                    'Bucket': bucket_name,
                    'Name': object_key
                }
            },
            MaxLabels=5,
            MinConfidence=75
        )

        analysis['labels'] = [
            {
                'name': label['Name'],
                'confidence': round(label['Confidence'], 2)
            }
            for label in labels_response.get('Labels', [])
        ]

        logger.info(f"Detected {len(analysis['labels'])} labels")

    except Exception as e:
        logger.error("Failed to detect labels", extra={"error": str(e)})

    return analysis

Lastly, the Lambda saves the results to the PropertyImagesTable DynamoDB table via the PynamoDB model:

def save_image_analysis(property_id: str, image_name: str, image_key: str, analysis: dict):
    """
    Save image metadata and analysis results to PropertyImageTable using PropertyImageModel.
    """
    try:
        # Create image record using Pynamodb model
        image_item = PropertyImageModel.create_image(
            property_id=property_id,
            image_name=image_name,
            image_key=image_key,
            labels=analysis.get('labels', []),
            analyzedAt=datetime.now()
        )

        # Save to DynamoDB
        image_item.save()

        logger.info("Image analysis saved to PropertyImageTable", extra={
            "property_id": property_id,
            "image_name": image_name,
            "image_key": image_key,
            "labels_count": len(analysis.get('labels', []))
        })

    except Exception as e:
        logger.error("Failed to save image analysis", extra={
            "error": str(e),
            "property_id": property_id,
            "image_name": image_name,
            "image_key": image_key
        }, exc_info=True)

Architecture Overview - AWS Bedrock Property Summarization

Here is a simple overview of the Summarization part:

The Lambda is available to the user via API Gateway, and when the user calls the URL, Lambda fetches all images from the PropertyImagesTable for the requested property and sends those labels to an AWS Bedrock model to get the summary.

We are using the amazon.nova-micro-v1:0 model, which is the cheapest model I found inside the model repository and it got the job done!

It’s worth mentioning that we won’t be opening this Lambda up to the Internet for now - this part of the backend is a part of the bigger project which will be covered in next blog posts and that’s why this part may seem simple! The following code will be enough to get you off the ground in any Bedrock integration!

Code Deep-Dive - AWS Bedrock Property Summarization

Here is the CDK code for generating necessary resources and adding necessary IAM permissions for accessing the Bedrock model:

# Add Bedrock permissions to the role
bedrock_lambda_role.add_to_policy(
    iam.PolicyStatement(
        effect=iam.Effect.ALLOW,
        actions=[
            "bedrock:InvokeModel",
        ],
        resources=[
            f"arn:aws:bedrock:*:{self.account}:inference-profile/eu.amazon.nova-micro-v1:0",
            "arn:aws:bedrock:*::foundation-model/amazon.nova-micro-v1:0"
        ],
    )
)

# Bedrock Summary Generation Lambda
summary_generation_code_dir = "./GetPropertySummaryHandler"

summary_generation_fn = _lambda.Function(
    self, "SummaryGenerationLambda",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="handler.handler",
    code=_lambda.Code.from_asset(summary_generation_code_dir),
    role=bedrock_lambda_role,
    environment={
        "PROPERTY_IMAGES_TABLE": self.property_images_table.table_name,
        "BEDROCK_MODEL_ID": "eu.amazon.nova-micro-v1:0",
        "ALLOWED_ORIGIN": "*",
    },
    timeout=Duration.seconds(30)
)

# Giving the SummaryGenerationLambda permissions to
# look into the DynamoDB tables
self.property_images_table.grant_read_data(summary_generation_fn)

Now, the Lambda code - to be completely honest, I expected this part to be way harder to configure, but it was really easy!

Building the request_body JSON was interesting, because inside the request_body.system.text , you can define how to customize the response of the model. The model was instructed to act as following:

You are an experienced real estate agent creating property listings. Write concise, appealing, and professional property descriptions that highlight key features and amenities based on the labels provided. Keep descriptions brief (up to 10 sentences) and focus on what makes the property attractive to potential buyers or renters. Use descriptive language but remain factual based on the provided information.

You can also configure the maximum response length via a field max_new_tokens which was set to 200 to keep the costs as low as possible. In addition, configuration fields like temperature and top_p were used - for more information about this, please see this blog post by clicking on this link.

Here is the code to get the response from the Bedrock model:

# Initialize AWS clients
BEDROCK_CLIENT = client("bedrock-runtime")
BEDROCK_MODEL_ID = os.environ.get("BEDROCK_MODEL_ID")

# Fetch labels from PropertyImagesTable using model
images = PropertyImageModel.get_property_images(property_id)

# Extract all unique labels from all images
all_labels = []
for image_item in images:
    for label in image_item.labels or []:
        if isinstance(label, dict):
            label_name = label.get('name', '')
            if label_name:
                all_labels.append(label_name)
        elif isinstance(label, str):
            all_labels.append(label)

property_labels_str = ", ".join(set(all_labels))

request_body = {
    "system": [
        {
            "text": "You are an experienced real estate agent creating property listings. Write concise, appealing, and professional property descriptions that highlight key features and amenities based on the labels provided. Keep descriptions brief (up to 10 sentences) and focus on what makes the property attractive to potential buyers or renters. Use descriptive language but remain factual based on the provided information."
        }
    ],
    "messages": [
        {
            "role": "user",
            "content": [
                {
                    "text": property_labels_str
                }
            ]
        }
    ],
    "inferenceConfig": {
        "max_new_tokens": 200,
        "temperature": 0.7,
        "top_p": 0.9
    }
}

response = BEDROCK_CLIENT.invoke_model(
    modelId=BEDROCK_MODEL_ID,
    contentType="application/json",
    accept="application/json",
    body=json.dumps(request_body)
)

Conclusion

This architecture cuts image-processing costs, improves backend reliability and scales with ease.

By combining AWS services like S3, Lambda, Rekognition and Bedrock, with the power of AWS CDK as the Infrastructure-as-Code solution, you’ve learned how to create an AI powered pipeline, which can be applied to ANY project.

In the following post, we will go over the rest of the Property stack, which is responsible for CRUD operations with best practices and best performance for fetching images from S3 and data from DynamoDB tables.

Thank you for reading, see you in the next post!