aws_utils_dynamodb

AWS DynamoDB utilities for Rust, providing a simplified interface for common DynamoDB operations.

Features

• Simple DynamoDB client creation with configurable endpoint
• Record operations (CRUD)
• Table management operations
• Stream-based pagination for scan and query operations
• CSV import functionality from S3
• Error handling with custom error types

Installation

Add this to your Cargo.toml:

[dependencies]
aws_utils_dynamodb = { path = "crates/dynamodb" }

Usage

Creating a Client

use aws_utils_dynamodb::{make_client, make_client_with_timeout_default, make_client_with_timeout};
use std::time::Duration;

// Create client with default timeout settings
let client = make_client_with_timeout_default(None).await;

// Create client with custom timeout settings
let client = make_client_with_timeout(
    None, // endpoint_url
    Some(Duration::from_secs(3100)), // connect_timeout
    Some(Duration::from_secs(60)),   // operation_timeout
    Some(Duration::from_secs(55)),   // operation_attempt_timeout
    Some(Duration::from_secs(50)),   // read_timeout
).await;

// Create client with custom endpoint and default timeout
let client = make_client_with_timeout_default(
    Some("http://localhost:8000".to_string())
).await;

// Create client without timeout configuration (legacy)
let client = make_client(None, None, None).await;

// Create client with custom endpoint and no timeout (legacy)
let client = make_client(Some("http://localhost:8000".to_string()), None, None).await;

Logging AWS Communication

make_client accepts an optional [SharedInterceptor]. By passing an interceptor that implements aws_sdk_dynamodb::config::Intercept, you can run custom logic — such as logging — every time the client communicates with AWS.

The interceptor below logs each request, response, and operation result. It uses the tracing crate, which is also what the AWS SDK uses internally.

use aws_utils_dynamodb::make_client;
use aws_sdk_dynamodb::config::{
    ConfigBag, Intercept, RuntimeComponents, SharedInterceptor,
    interceptors::{
        AfterDeserializationInterceptorContextRef, BeforeDeserializationInterceptorContextRef,
        BeforeTransmitInterceptorContextRef,
    },
};

type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;

#[derive(Debug, Clone)]
struct LoggingInterceptor;

impl Intercept for LoggingInterceptor {
    fn name(&self) -> &'static str {
        "DynamoDbLoggingInterceptor"
    }

    // Called just before each HTTP request is sent (once per retry attempt).
    fn read_before_transmit(
        &self,
        context: &BeforeTransmitInterceptorContextRef<'_>,
        _runtime_components: &RuntimeComponents,
        _cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        let request = context.request();
        tracing::info!(
            method = %request.method(),
            uri = %request.uri(),
            "DynamoDB -> AWS request"
        );
        Ok(())
    }

    // Called right after each HTTP response is received.
    fn read_before_deserialization(
        &self,
        context: &BeforeDeserializationInterceptorContextRef<'_>,
        _runtime_components: &RuntimeComponents,
        _cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        let response = context.response();
        tracing::info!(status = %response.status(), "AWS -> DynamoDB response");
        Ok(())
    }

    // Called once when the operation completes (after retries), with success or error.
    fn read_after_deserialization(
        &self,
        context: &AfterDeserializationInterceptorContextRef<'_>,
        _runtime_components: &RuntimeComponents,
        _cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        match context.output_or_error() {
            Ok(_) => tracing::info!("DynamoDB operation succeeded"),
            Err(err) => tracing::warn!(error = %err, "DynamoDB operation failed"),
        }
        Ok(())
    }
}

# async fn run() {
// Pass the interceptor as the third argument.
let client = make_client(None, None, Some(SharedInterceptor::new(LoggingInterceptor))).await;
# }

tracing does not emit anything until a subscriber is initialized. Set one up once in your application (for example with tracing-subscriber) and control verbosity with RUST_LOG:

// Add `tracing-subscriber` to your dependencies.
tracing_subscriber::fmt()
    .with_env_filter(
        tracing_subscriber::EnvFilter::try_from_default_env()
            .unwrap_or_else(|_| "info".into()),
    )
    .init();

Example output (RUST_LOG=info):

INFO DynamoDbLoggingInterceptor: DynamoDB -> AWS request method=POST uri=https://dynamodb.ap-northeast-1.amazonaws.com/
INFO DynamoDbLoggingInterceptor: AWS -> DynamoDB response status=200
INFO DynamoDbLoggingInterceptor: DynamoDB operation succeeded

Record Operations

use aws_utils_dynamodb::record::{get_item, put_item, update_item, delete_item, scan_all, query_all, query};
use aws_sdk_dynamodb::types::{AttributeValue, ReturnValue};
use std::collections::HashMap;

// Get an item
let mut key = HashMap::new();
key.insert("id".to_string(), AttributeValue::S("123".to_string()));
let item = get_item(&client, "my_table", key).await?;

// Put an item
let mut item = HashMap::new();
item.insert("id".to_string(), AttributeValue::S("123".to_string()));
item.insert("name".to_string(), AttributeValue::S("John".to_string()));
let output = put_item(&client, "my_table", item, None, None, None, None).await?;

// Update an item
let mut key = HashMap::new();
key.insert("id".to_string(), AttributeValue::S("123".to_string()));
let output = update_item(
    &client,
    "my_table",
    key,
    "SET #name = :name",
    None,
    Some(HashMap::from([("#name".to_string(), "name".to_string())])),
    Some(HashMap::from([(":name".to_string(), AttributeValue::S("Jane".to_string()))])),
    Some(ReturnValue::AllNew)
).await?;

// Delete an item
let mut key = HashMap::new();
key.insert("id".to_string(), AttributeValue::S("123".to_string()));
let output = delete_item(&client, "my_table", key, None, None, None, None).await?;

// Scan all items
let items = scan_all(&client, "my_table", None, None, None, None).await?;

// Query items (all)
let items = query_all(
    &client,
    "my_table",
    None,
    Some("id = :id"),
    None,
    None,
    Some(HashMap::from([(":id".to_string(), AttributeValue::S("123".to_string()))]))
).await?;

// Query items (with limit, no pagination)
let items = query(
    &client,
    "my_table",
    None,          // index_name
    Some("id = :id"),
    None,          // filter_expression
    None,          // expression_attribute_names
    Some(HashMap::from([(":id".to_string(), AttributeValue::S("123".to_string()))])),
    None,          // consistent_read
    None::<String>, // projection_expression
    None::<Vec<String>>, // attributes_to_get
    Some(10),      // limit
).await?;

Table Operations

use aws_utils_dynamodb::table::{create_table, delete_table, describe_table, get_capacity, set_capacity, TableType};
use aws_sdk_dynamodb::types::{AttributeDefinition, ScalarAttributeType};

// Create a table with on-demand billing
let attrs = vec![
    AttributeDefinition::builder()
        .attribute_name("id")
        .attribute_type(ScalarAttributeType::S)
        .build()?,
];
let output = create_table(
    &client,
    "my_table",
    "id",
    None::<String>,
    TableType::OnDemand,
    attrs,
    None
).await?;

// Create a table with provisioned capacity
let output = create_table(
    &client,
    "my_table",
    "id",
    Some("timestamp"),
    TableType::Provisioned(5, 5), // 5 RCU, 5 WCU
    attrs,
    None
).await?;

// Delete a table
let output = delete_table(&client, "my_table").await?;

// Get table capacity
let (read_units, write_units) = get_capacity(&client, "my_table").await?;

// Update table capacity
let output = set_capacity(&client, "my_table", 10, 10).await?;

CSV Import from S3

use aws_utils_dynamodb::csv::import_table;
use aws_utils_dynamodb::table::TableType;
use aws_sdk_dynamodb::types::{AttributeDefinition, ScalarAttributeType};

// Import CSV data from S3 to a new DynamoDB table
let attrs = vec![
    AttributeDefinition::builder()
        .attribute_name("id")
        .attribute_type(ScalarAttributeType::S)
        .build()?,
];

import_table(
    &client,
    "my-bucket",
    "data/users.csv",
    Some(","), // delimiter
    Some(vec!["id".to_string(), "name".to_string(), "email".to_string()]), // headers
    "imported_users_table",
    "id", // hash key
    None::<String>, // no sort key
    attrs,
    TableType::OnDemand
).await?;

Stream Operations

For handling large datasets, use stream-based operations:

use aws_utils_dynamodb::record::{scan_stream, query_stream};
use futures_util::TryStreamExt;

// Scan with streaming
let stream = scan_stream(&client, "my_table", None, None, None, None);
futures_util::pin_mut!(stream);
while let Some(item) = stream.try_next().await? {
    // Process each item
    println!("{:?}", item);
}

// Query with streaming
let stream = query_stream(
    &client,
    "my_table",
    None,
    Some("id = :id"),
    None,
    None,
    Some(HashMap::from([(":id".to_string(), AttributeValue::S("123".to_string()))]))
);
futures_util::pin_mut!(stream);
while let Some(item) = stream.try_next().await? {
    // Process each item
    println!("{:?}", item);
}

Error Handling

The crate provides a custom Error type that wraps AWS SDK errors and includes common error cases:

• NotFound - Item not found
• ValidationError - Invalid parameters or state
• Invalid - Invalid response from AWS
• AwsSdkError - AWS SDK specific errors

Environment Variables

The client uses the AWS SDK's default credential chain, which checks for credentials in the following order:

• Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION)
• ECS task role (for Fargate/ECS)
• EC2 instance profile
• AWS credentials file
• Other configured credential providers

License

This project is part of the utilities.aws-utils workspace.