# 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`:
```toml
[dependencies]
aws_utils_dynamodb = { path = "crates/dynamodb" }
```
## Usage
### Creating a Client
```rust
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`](https://crates.io/crates/tracing) crate, which is also what the AWS SDK uses
internally.
```rust
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`:
```rust
// 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`):
```text
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
```rust
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
```rust
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
```rust
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:
```rust
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.