DynamoDB examples using SDK for Rust - AWS SDK for Rust

DynamoDB examples using SDK for Rust

The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Rust with DynamoDB.

Actions are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios and cross-service examples.

Scenarios are code examples that show you how to accomplish a specific task by calling multiple functions within the same service.

Each example includes a link to GitHub, where you can find instructions on how to set up and run the code in context.

Actions

The following code example shows how to use CreateTable.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

pub async fn create_table( client: &Client, table: &str, key: &str, ) -> Result<CreateTableOutput, Error> { let a_name: String = key.into(); let table_name: String = table.into(); let ad = AttributeDefinition::builder() .attribute_name(&a_name) .attribute_type(ScalarAttributeType::S) .build() .map_err(Error::BuildError)?; let ks = KeySchemaElement::builder() .attribute_name(&a_name) .key_type(KeyType::Hash) .build() .map_err(Error::BuildError)?; let pt = ProvisionedThroughput::builder() .read_capacity_units(10) .write_capacity_units(5) .build() .map_err(Error::BuildError)?; let create_table_response = client .create_table() .table_name(table_name) .key_schema(ks) .attribute_definitions(ad) .provisioned_throughput(pt) .send() .await; match create_table_response { Ok(out) => { println!("Added table {} with key {}", table, key); Ok(out) } Err(e) => { eprintln!("Got an error creating table:"); eprintln!("{}", e); Err(Error::unhandled(e)) } } }
  • For API details, see CreateTable in AWS SDK for Rust API reference.

The following code example shows how to use DeleteItem.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

pub async fn delete_item( client: &Client, table: &str, key: &str, value: &str, ) -> Result<DeleteItemOutput, Error> { match client .delete_item() .table_name(table) .key(key, AttributeValue::S(value.into())) .send() .await { Ok(out) => { println!("Deleted item from table"); Ok(out) } Err(e) => Err(Error::unhandled(e)), } }
  • For API details, see DeleteItem in AWS SDK for Rust API reference.

The following code example shows how to use DeleteTable.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

pub async fn delete_table(client: &Client, table: &str) -> Result<DeleteTableOutput, Error> { let resp = client.delete_table().table_name(table).send().await; match resp { Ok(out) => { println!("Deleted table"); Ok(out) } Err(e) => Err(Error::Unhandled(e.into())), } }
  • For API details, see DeleteTable in AWS SDK for Rust API reference.

The following code example shows how to use ListTables.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

pub async fn list_tables(client: &Client) -> Result<Vec<String>, Error> { let paginator = client.list_tables().into_paginator().items().send(); let table_names = paginator.collect::<Result<Vec<_>, _>>().await?; println!("Tables:"); for name in &table_names { println!(" {}", name); } println!("Found {} tables", table_names.len()); Ok(table_names) }

Determine whether table exists.

pub async fn table_exists(client: &Client, table: &str) -> Result<bool, Error> { debug!("Checking for table: {table}"); let table_list = client.list_tables().send().await; match table_list { Ok(list) => Ok(list.table_names().contains(&table.into())), Err(e) => Err(e.into()), } }
  • For API details, see ListTables in AWS SDK for Rust API reference.

The following code example shows how to use PutItem.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

pub async fn add_item(client: &Client, item: Item, table: &String) -> Result<ItemOut, Error> { let user_av = AttributeValue::S(item.username); let type_av = AttributeValue::S(item.p_type); let age_av = AttributeValue::S(item.age); let first_av = AttributeValue::S(item.first); let last_av = AttributeValue::S(item.last); let request = client .put_item() .table_name(table) .item("username", user_av) .item("account_type", type_av) .item("age", age_av) .item("first_name", first_av) .item("last_name", last_av); println!("Executing request [{request:?}] to add item..."); let resp = request.send().await?; let attributes = resp.attributes().unwrap(); let username = attributes.get("username").cloned(); let first_name = attributes.get("first_name").cloned(); let last_name = attributes.get("last_name").cloned(); let age = attributes.get("age").cloned(); let p_type = attributes.get("p_type").cloned(); println!( "Added user {:?}, {:?} {:?}, age {:?} as {:?} user", username, first_name, last_name, age, p_type ); Ok(ItemOut { p_type, age, username, first_name, last_name, }) }
  • For API details, see PutItem in AWS SDK for Rust API reference.

The following code example shows how to use Query.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Find the movies made in the specified year.

pub async fn movies_in_year( client: &Client, table_name: &str, year: u16, ) -> Result<Vec<Movie>, MovieError> { let results = client .query() .table_name(table_name) .key_condition_expression("#yr = :yyyy") .expression_attribute_names("#yr", "year") .expression_attribute_values(":yyyy", AttributeValue::N(year.to_string())) .send() .await?; if let Some(items) = results.items { let movies = items.iter().map(|v| v.into()).collect(); Ok(movies) } else { Ok(vec![]) } }
  • For API details, see Query in AWS SDK for Rust API reference.

The following code example shows how to use Scan.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

pub async fn list_items(client: &Client, table: &str, page_size: Option<i32>) -> Result<(), Error> { let page_size = page_size.unwrap_or(10); let items: Result<Vec<_>, _> = client .scan() .table_name(table) .limit(page_size) .into_paginator() .items() .send() .collect() .await; println!("Items in table (up to {page_size}):"); for item in items? { println!(" {:?}", item); } Ok(()) }
  • For API details, see Scan in AWS SDK for Rust API reference.

Scenarios

The following code example shows how to:

  • Get an item by running a SELECT statement.

  • Add an item by running an INSERT statement.

  • Update an item by running an UPDATE statement.

  • Delete an item by running a DELETE statement.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

async fn make_table( client: &Client, table: &str, key: &str, ) -> Result<(), SdkError<CreateTableError>> { let ad = AttributeDefinition::builder() .attribute_name(key) .attribute_type(ScalarAttributeType::S) .build() .expect("creating AttributeDefinition"); let ks = KeySchemaElement::builder() .attribute_name(key) .key_type(KeyType::Hash) .build() .expect("creating KeySchemaElement"); let pt = ProvisionedThroughput::builder() .read_capacity_units(10) .write_capacity_units(5) .build() .expect("creating ProvisionedThroughput"); match client .create_table() .table_name(table) .key_schema(ks) .attribute_definitions(ad) .provisioned_throughput(pt) .send() .await { Ok(_) => Ok(()), Err(e) => Err(e), } } async fn add_item(client: &Client, item: Item) -> Result<(), SdkError<ExecuteStatementError>> { match client .execute_statement() .statement(format!( r#"INSERT INTO "{}" VALUE {{ "{}": ?, "acount_type": ?, "age": ?, "first_name": ?, "last_name": ? }} "#, item.table, item.key )) .set_parameters(Some(vec![ AttributeValue::S(item.utype), AttributeValue::S(item.age), AttributeValue::S(item.first_name), AttributeValue::S(item.last_name), ])) .send() .await { Ok(_) => Ok(()), Err(e) => Err(e), } } async fn query_item(client: &Client, item: Item) -> bool { match client .execute_statement() .statement(format!( r#"SELECT * FROM "{}" WHERE "{}" = ?"#, item.table, item.key )) .set_parameters(Some(vec![AttributeValue::S(item.value)])) .send() .await { Ok(resp) => { if !resp.items().is_empty() { println!("Found a matching entry in the table:"); println!("{:?}", resp.items.unwrap_or_default().pop()); true } else { println!("Did not find a match."); false } } Err(e) => { println!("Got an error querying table:"); println!("{}", e); process::exit(1); } } } async fn remove_item(client: &Client, table: &str, key: &str, value: String) -> Result<(), Error> { client .execute_statement() .statement(format!(r#"DELETE FROM "{table}" WHERE "{key}" = ?"#)) .set_parameters(Some(vec![AttributeValue::S(value)])) .send() .await?; println!("Deleted item."); Ok(()) } async fn remove_table(client: &Client, table: &str) -> Result<(), Error> { client.delete_table().table_name(table).send().await?; Ok(()) }

Serverless examples

The following code example shows how to implement a Lambda function that receives an event triggered by receiving records from a DynamoDB stream. The function retrieves the DynamoDB payload and logs the record contents.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Consuming a DynamoDB event with Lambda using Rust.

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 use lambda_runtime::{service_fn, tracing, Error, LambdaEvent}; use aws_lambda_events::{ event::dynamodb::{Event, EventRecord}, }; // Built with the following dependencies: //lambda_runtime = "0.11.1" //serde_json = "1.0" //tokio = { version = "1", features = ["macros"] } //tracing = { version = "0.1", features = ["log"] } //tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] } //aws_lambda_events = "0.15.0" async fn function_handler(event: LambdaEvent<Event>) ->Result<(), Error> { let records = &event.payload.records; tracing::info!("event payload: {:?}",records); if records.is_empty() { tracing::info!("No records found. Exiting."); return Ok(()); } for record in records{ log_dynamo_dbrecord(record); } tracing::info!("Dynamo db records processed"); // Prepare the response Ok(()) } fn log_dynamo_dbrecord(record: &EventRecord)-> Result<(), Error>{ tracing::info!("EventId: {}", record.event_id); tracing::info!("EventName: {}", record.event_name); tracing::info!("DynamoDB Record: {:?}", record.change ); Ok(()) } #[tokio::main] async fn main() -> Result<(), Error> { tracing_subscriber::fmt() .with_max_level(tracing::Level::INFO) .with_target(false) .without_time() .init(); let func = service_fn(function_handler); lambda_runtime::run(func).await?; Ok(()) }

The following code example shows how to implement partial batch response for Lambda functions that receive events from a DynamoDB stream. The function reports the batch item failures in the response, signaling to Lambda to retry those messages later.

SDK for Rust
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Reporting DynamoDB batch item failures with Lambda using Rust.

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 use aws_lambda_events::{ event::dynamodb::{Event, EventRecord, StreamRecord}, streams::{DynamoDbBatchItemFailure, DynamoDbEventResponse}, }; use lambda_runtime::{run, service_fn, Error, LambdaEvent}; /// Process the stream record fn process_record(record: &EventRecord) -> Result<(), Error> { let stream_record: &StreamRecord = &record.change; // process your stream record here... tracing::info!("Data: {:?}", stream_record); Ok(()) } /// Main Lambda handler here... async fn function_handler(event: LambdaEvent<Event>) -> Result<DynamoDbEventResponse, Error> { let mut response = DynamoDbEventResponse { batch_item_failures: vec![], }; let records = &event.payload.records; if records.is_empty() { tracing::info!("No records found. Exiting."); return Ok(response); } for record in records { tracing::info!("EventId: {}", record.event_id); // Couldn't find a sequence number if record.change.sequence_number.is_none() { response.batch_item_failures.push(DynamoDbBatchItemFailure { item_identifier: Some("".to_string()), }); return Ok(response); } // Process your record here... if process_record(record).is_err() { response.batch_item_failures.push(DynamoDbBatchItemFailure { item_identifier: record.change.sequence_number.clone(), }); /* Since we are working with streams, we can return the failed item immediately. Lambda will immediately begin to retry processing from this failed item onwards. */ return Ok(response); } } tracing::info!("Successfully processed {} record(s)", records.len()); Ok(response) } #[tokio::main] async fn main() -> Result<(), Error> { tracing_subscriber::fmt() .with_max_level(tracing::Level::INFO) // disable printing the name of the module in every log line. .with_target(false) // disabling time is handy because CloudWatch will add the ingestion time. .without_time() .init(); run(service_fn(function_handler)).await }