toasty-driver-dynamodb 0.3.0

Amazon DynamoDB driver for Toasty
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

use super::{
    Connection, ExprAttrs, Result, Schema, ddb_expression, deserialize_ddb_cursor, item_to_record,
    operation, serialize_ddb_cursor, stmt,
};
use std::sync::Arc;
use toasty_core::{driver::ExecResponse, stmt::ExprContext};

impl Connection {
    pub(crate) async fn exec_query_pk(
        &mut self,
        schema: &Arc<Schema>,
        op: operation::QueryPk,
    ) -> Result<ExecResponse> {
        let table = schema.db.table(op.table);
        let cx = ExprContext::new_with_target(&schema.db, table);

        let mut expr_attrs = ExprAttrs::default();

        // When querying an index, use index filter logic (not primary key logic)
        let is_primary_key = op.index.is_none();
        let key_expression = ddb_expression(&cx, &mut expr_attrs, is_primary_key, &op.pk_filter);

        let filter_expression = op
            .filter
            .as_ref()
            .map(|expr| ddb_expression(&cx, &mut expr_attrs, false, expr));

        // Build the query based on whether we're querying primary key or an index
        let result = if let Some(index_id) = op.index {
            let index = schema.db.index(index_id);

            if index.unique {
                use toasty_core::Error;
                let err = Error::from_args(format_args!(
                    "Unique index {} doesn't have fields.",
                    index.name
                ));
                Err(err)
            } else {
                tracing::trace!(table_name = %table.name, index_name = %index.name, "querying secondary index");
                let mut query = self
                    .client
                    .query()
                    .table_name(&table.name)
                    .index_name(&index.name)
                    .key_condition_expression(key_expression)
                    .set_filter_expression(filter_expression)
                    .set_expression_attribute_names(Some(expr_attrs.attr_names))
                    .set_expression_attribute_values(Some(expr_attrs.attr_values));

                if let Some(limit) = op.limit {
                    query = query.limit(limit as i32);
                }
                if let Some(ref cursor_value) = op.cursor {
                    query =
                        query.set_exclusive_start_key(Some(deserialize_ddb_cursor(cursor_value)));
                }
                if let Some(ref direction) = op.order {
                    query = query.scan_index_forward(*direction == stmt::Direction::Asc);
                }

                query
                    .send()
                    .await
                    .map_err(toasty_core::Error::driver_operation_failed)
            }
        } else {
            tracing::trace!(table_name = %table.name, "querying primary key");
            let mut query = self
                .client
                .query()
                .table_name(&table.name)
                .key_condition_expression(key_expression)
                .set_filter_expression(filter_expression)
                .set_expression_attribute_names(Some(expr_attrs.attr_names))
                .set_expression_attribute_values(Some(expr_attrs.attr_values));

            if let Some(limit) = op.limit {
                query = query.limit(limit as i32);
            }
            if let Some(ref cursor_value) = op.cursor {
                query = query.set_exclusive_start_key(Some(deserialize_ddb_cursor(cursor_value)));
            }
            if let Some(ref direction) = op.order {
                query = query.scan_index_forward(*direction == stmt::Direction::Asc);
            }

            query
                .send()
                .await
                .map_err(toasty_core::Error::driver_operation_failed)
        };

        let schema = schema.clone();
        let res = result?;

        // Capture LastEvaluatedKey for pagination
        let cursor = res.last_evaluated_key.as_ref().map(serialize_ddb_cursor);

        let rows = stmt::ValueStream::from_iter(res.items.into_iter().flatten().map(move |item| {
            item_to_record(
                &item,
                op.select
                    .iter()
                    .map(|column_id| schema.db.column(*column_id)),
            )
        }));

        Ok(ExecResponse {
            values: toasty_core::driver::Rows::Stream(rows),
            next_cursor: cursor,
            prev_cursor: None,
        })
    }
}