aws-expect 1.4.0

Declarative expect/wait syntax for AWS services (S3, DynamoDB, and more)

AWS Expect

Declarative, Pythonic waiters for AWS services using boto3. Wait for S3 objects, DynamoDB items/tables, SQS messages/events, and Lambda functions to reach an expected state — with content matching, stop_when predicates for early abort, structured Expected:/Actual: error messages, and parallel execution via expect_all / expect_any.

Features

• Declarative syntax: expect_s3(obj).to_exist(timeout=30)
• Content matching: Wait for S3 body JSON or DynamoDB item attributes to match expected values
• Smart polling: stop_when predicates abort early when further polling is pointless
• Richer errors: Structured Expected:/Actual: sections in timeout error messages
• Native boto3 waiters: Uses AWS's built-in waiter infrastructure where available
• Testing-friendly: Perfect for integration tests and CI/CD pipelines
• Resource-based: Works with boto3 resource objects (S3, DynamoDB, SQS) and client (Lambda)
• Flexible timeouts: Configure both timeout and poll intervals
• Parallel waiting: expect_all() / expect_any() run multiple expectations concurrently, accepting both plain callables and (fn, args, kwargs) tuples
• Zero-boilerplate tuples: (fn, args, kwargs) eliminates lambda: wrappers at call sites

Installation

pip install aws-expect

Or with uv:

uv add aws-expect

Quick Start

S3 Object Waiting

import boto3
from aws_expect import expect_s3, S3WaitTimeoutError

s3 = boto3.resource("s3")
obj = s3.Object("my-bucket", "report.csv")

metadata = expect_s3(obj).to_exist(timeout=30, poll_interval=5)
print(f"Object exists! Size: {metadata['ContentLength']} bytes")

expect_s3(obj).to_not_exist(timeout=10, poll_interval=2)

# Wait for object body to be valid JSON that deep-matches expected content
from aws_expect import S3ContentWaitTimeoutError

body = expect_s3(obj).to_have_content({"status": "shipped"}, timeout=30)

# Assert object body does NOT match after a delay
from aws_expect import S3UnexpectedContentError

expect_s3(obj).to_not_have_content({"status": "cancelled"}, delay=5)

# Abort early with stop_when predicate
from aws_expect import StopConditionMetError

body = expect_s3(obj).to_exist(
    entries={"status": "shipped"},
    stop_when=lambda state: state.get("status") == "cancelled",
    timeout=60,
)

DynamoDB Item Waiting

import boto3
from aws_expect import expect_dynamodb_item, DynamoDBWaitTimeoutError

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("orders")

item = expect_dynamodb_item(table).to_exist(
    key={"pk": "order-123"},
    entries={"status": "shipped"},
    timeout=60,
    poll_interval=5,
)

expect_dynamodb_item(table).to_not_exist(key={"pk": "order-123"}, timeout=10)

# Scan table for an item matching entries
from aws_expect import DynamoDBFindItemTimeoutError

item = expect_dynamodb_item(table).to_find_item(
    entries={"status": "pending"},
    timeout=30,
)

# Assert no matching item exists after a delay
from aws_expect import DynamoDBUnexpectedItemError

expect_dynamodb_item(table).to_not_find_item({"status": "cancelled"}, delay=5)

# Abort scan early with stop_when predicate
item = expect_dynamodb_item(table).to_find_item(
    entries={"status": "pending"},
    stop_when=lambda item: item.get("status") == "failed",
    timeout=60,
)

DynamoDB Table Waiting

from aws_expect import expect_dynamodb_table

dynamodb = boto3.resource("dynamodb")

description = expect_dynamodb_table(dynamodb, "orders").to_exist(timeout=30)
expect_dynamodb_table(dynamodb, "orders").to_not_exist(timeout=30)

SQS Message Waiting

import boto3
from aws_expect import expect_sqs, SQSWaitTimeoutError

sqs = boto3.resource("sqs")
queue = sqs.Queue("https://sqs.us-east-1.amazonaws.com/123456789/my-queue")

# Wait for a plain-text message (non-destructive)
msg = expect_sqs(queue).to_have_message("order-confirmed", timeout=30)

# Wait and consume (delete) the matching message
msg = expect_sqs(queue).to_consume_message("order-confirmed", timeout=30)

# Assert message is absent after a delay
expect_sqs(queue).to_not_have_message("order-confirmed", delay=5)

SQS JSON Event Waiting

# Wait for a JSON message that deep-matches the expected event (non-destructive)
msg = expect_sqs(queue).to_have_event({"type": "ORDER_CREATED", "orderId": "123"}, timeout=30)

# Wait and consume (delete) the matching event
msg = expect_sqs(queue).to_consume_event({"type": "ORDER_CREATED"}, timeout=30)

# Assert no matching event is present after a delay
expect_sqs(queue).to_not_have_event({"type": "ORDER_CREATED"}, delay=5)

Lambda Function Waiting

import boto3
from aws_expect import expect_lambda, LambdaWaitTimeoutError

lambda_client = boto3.client("lambda")

# Wait for function to exist / be deleted
response = expect_lambda(lambda_client).to_exist("my-function", timeout=30)
expect_lambda(lambda_client).to_not_exist("my-function", timeout=30)

# Wait for function to reach Active state after deployment
response = expect_lambda(lambda_client).to_be_active("my-function", timeout=60)

# Wait for a function update to complete
response = expect_lambda(lambda_client).to_be_updated("my-function", timeout=60)

# Wait until the function can be invoked and returns expected output
payload = expect_lambda(lambda_client).to_be_invocable(
    "my-function",
    payload={"key": "value"},
    entries={"statusCode": 200},
    timeout=30,
)

# Invoke once and assert the response (not a waiter — raises immediately on mismatch)
from aws_expect import LambdaResponseMismatchError

result = expect_lambda(lambda_client).to_respond_with(
    "my-function",
    status_code=200,
    body={"message": "hello"},
    payload={"key": "value"},
)

Parallel Waiting

from aws_expect import expect_all, expect_s3, expect_dynamodb_item

# Wait for ALL expectations to succeed (raises if any times out)
results = expect_all([
    lambda: expect_s3(obj).to_exist(timeout=30),
    lambda: expect_dynamodb_item(table).to_exist(key={"pk": "order-123"}, timeout=30),
])

# Wait for the FIRST expectation to succeed (returns its result)
from aws_expect import expect_any, expect_dynamodb_item

result = expect_any([
    lambda: expect_dynamodb_item(table_a).to_exist(key={"pk": "u1"}, timeout=30),
    lambda: expect_dynamodb_item(table_b).to_exist(key={"pk": "u1"}, timeout=30),
])

# Tuple form — pass (callable, args, kwargs) to skip lambda: boilerplate
results = expect_all([
    (expect_dynamodb_item(table_a).to_exist, ({"pk": "a"}, 30, 1), {}),
    (expect_dynamodb_item(table_b).to_exist, (), {"key": {"pk": "b"}, "timeout": 30}),
])

result = expect_any([
    (expect_dynamodb_item(table_a).to_exist, ({"pk": "u1"}, 30, 1), {}),
    (expect_dynamodb_item(table_b).to_exist, ({"pk": "u1"}, 30, 1), {}),
])

Catching Any Timeout

All service-specific exceptions inherit from WaitTimeoutError:

from aws_expect import WaitTimeoutError

try:
    expect_s3(obj).to_exist(timeout=30)
except WaitTimeoutError:
    print("Timed out waiting for resource")

API Reference

Factory Functions

Function	Description
expect_s3(s3_object)	Creates S3ObjectExpectation
expect_dynamodb_item(table)	Creates DynamoDBItemExpectation
expect_dynamodb_table(dynamodb, table_name)	Creates DynamoDBTableExpectation
expect_sqs(queue)	Creates SQSQueueExpectation
expect_lambda(lambda_client)	Creates LambdaFunctionExpectation
expect_all(expectations)	Runs expectations concurrently; accepts callables or (fn, args, kwargs) tuples; returns all results or raises
expect_any(expectations)	Runs expectations concurrently; accepts callables or (fn, args, kwargs) tuples; returns first to succeed

S3 (S3ObjectExpectation)

Method	Description
to_exist(timeout, poll_interval, entries, *, stop_when)	Wait for object to exist; optionally match metadata entries and abort early via stop_when
to_not_exist(timeout, poll_interval)	Wait for object to be deleted
to_have_content(entries, timeout, poll_interval)	Wait until object body is valid JSON deep-matching entries
to_not_have_content(entries, delay)	Assert object body does not deep-match entries after delay

DynamoDB Item (DynamoDBItemExpectation)

Method	Description
to_exist(key, timeout, poll_interval, entries, *, stop_when)	Wait for item to exist; optionally match attribute entries and abort early via stop_when
to_not_exist(key, timeout, poll_interval)	Wait for item to be deleted
to_be_empty(timeout, poll_interval)	Wait for table to have no items
to_be_not_empty(timeout, poll_interval)	Wait for table to have at least one item
to_have_numeric_value_close_to(key, field, value, delta, timeout, poll_interval)	Wait for a numeric field to be within delta of value
to_find_item(entries, timeout, poll_interval, *, stop_when)	Scan table until at least one item subset-matches entries; abort via stop_when
to_not_find_item(entries, delay)	Assert no item matches entries after delay

DynamoDB Table (DynamoDBTableExpectation)

Method	Description
to_exist(timeout, poll_interval)	Wait for table to exist (Active state)
to_not_exist(timeout, poll_interval)	Wait for table to be deleted

SQS (SQSQueueExpectation)

Method	Description
to_have_message(body, timeout, poll_interval)	Wait for exact-body message (non-destructive)
to_consume_message(body, timeout, poll_interval)	Wait and delete matching message
to_not_have_message(body, delay)	Assert message absent after delay
to_have_event(event, timeout, poll_interval)	Wait for JSON subset match (non-destructive)
to_consume_event(event, timeout, poll_interval)	Wait and delete matching JSON event
to_not_have_event(event, delay)	Assert JSON event absent after delay

Lambda (LambdaFunctionExpectation)

Method	Description
to_exist(function_name, timeout, poll_interval)	Wait for function to exist
to_not_exist(function_name, timeout, poll_interval)	Wait for function to be deleted
to_be_active(function_name, timeout, poll_interval)	Wait for State == "Active"
to_be_updated(function_name, timeout, poll_interval)	Wait for LastUpdateStatus == "Successful"
to_be_invocable(function_name, timeout, poll_interval, payload, entries)	Wait until invocation succeeds; optionally match response payload entries
to_respond_with(function_name, status_code, body, payload)	Invoke once and assert statusCode and/or JSON body (not a waiter)

Exceptions

All timeout exceptions inherit from WaitTimeoutError:

Exception	Raised by
S3WaitTimeoutError	S3 existence methods
S3ContentWaitTimeoutError	to_have_content
S3UnexpectedContentError	to_not_have_content (not a timeout)
DynamoDBWaitTimeoutError	DynamoDB methods
DynamoDBFindItemTimeoutError	to_find_item
DynamoDBUnexpectedItemError	to_not_find_item (not a timeout)
DynamoDBNonNumericFieldError	to_have_numeric_value_close_to (not a timeout)
SQSWaitTimeoutError	SQS string-body methods
SQSEventWaitTimeoutError	SQS JSON event methods
SQSUnexpectedMessageError	to_not_have_message (not a timeout)
SQSUnexpectedEventError	to_not_have_event (not a timeout)
LambdaWaitTimeoutError	Lambda methods
LambdaInvocableTimeoutError	to_be_invocable (with entries)
LambdaResponseMismatchError	to_respond_with (not a timeout)
StopConditionMetError	stop_when predicate returns True (not a timeout)
StopConditionError	stop_when predicate raises an exception (not a timeout)
AggregateWaitTimeoutError	expect_all, expect_any

Development

Setup

git clone https://github.com/PhishStick-hub/aws-expect
cd aws-expect
uv sync --all-groups

Running Tests

Tests use testcontainers and LocalStack for real AWS API simulation:

docker info
uv run pytest tests/ -v

License

MIT License - see LICENSE file for details.

Author

Ivan Shcherbenko

Credits

Built with:

• boto3 — AWS SDK for Python
• testcontainers-python — Testing with real services
• LocalStack — Local AWS cloud stack