Add ConsistentRead support for GetItem, Query, and BatchGetItems operations

README.md

@@ -163,6 +163,28 @@ const products = await dynamoHelper.query<ProductModel>({
 });
 ```
 
+#### Consistent Read
+
+By default, DynamoDB uses eventually consistent reads. To use strongly consistent reads, pass `true` as the third parameter to the query method:
+
+```typescript
+// Perform a strongly consistent read
+const products = await dynamoHelper.query<ProductModel>(
+  {
+    where: {
+      pk: 'org_uuid',
+      sk: {
+        beginsWith: 'product_',
+      },
+    },
+  },
+  undefined,  // indexName
+  true        // consistentRead
+);
+```
+
+**Note:** Consistent reads are only supported on the base table and local secondary indexes, not on global secondary indexes. Consistent reads consume twice the read capacity units compared to eventually consistent reads.
+
 ### pginated based query (cursor)
 
 - DynamoDB official docs - [Paginate table query result](https://docs.amazonaws.cn/en_us/amazondynamodb/latest/developerguide/Query.Pagination.html) and [Query](https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_Query.html)
@@ -242,6 +264,21 @@ await dynamoHelper.getItem<ProductModel>({ id: 'product_xxx' }, [
 ]);
 ```
 
+#### Consistent Read
+
+To use strongly consistent reads with `getItem`, pass `true` as the third parameter:
+
+```typescript
+// Get item with strongly consistent read
+const product = await dynamoHelper.getItem<ProductModel>(
+  { pk: 'org_uuid', sk: 'product_xxx' },
+  ['id', 'name', 'price'],
+  true  // consistentRead
+);
+```
+
+**Note:** Consistent reads consume twice the read capacity units compared to eventually consistent reads.
+
 ### batchGetItems
 
 Fetch many items using pk and sk combination
@@ -256,6 +293,24 @@ const products = await dynamoHelper.batchGetItems<ProductModel>([
 ]);
 ```
 
+#### Consistent Read
+
+To use strongly consistent reads with `batchGetItems`, pass `true` as the third parameter:
+
+```typescript
+// Batch get items with strongly consistent read
+const products = await dynamoHelper.batchGetItems<ProductModel>(
+  [
+    { pk: 'org_uuid', sk: 'product_1' },
+    { pk: 'org_uuid', sk: 'product_2' },
+  ],
+  ['id', 'name', 'price'],  // fields to project
+  true  // consistentRead
+);
+```
+
+**Note:** Consistent reads consume twice the read capacity units compared to eventually consistent reads.
+
 ### exists
 
 Check if an item exists in the database with the keys provided. Returns a boolean value

src/DynamoHelper.ts

@@ -50,30 +50,34 @@ export class DynamoHelper {
   async query<T extends AnyObject>(
     filter: Filter<T>,
     indexName?: string,
+    consistentRead?: boolean,
   ): Promise<Array<T>> {
-    return query(this.dbClient, this.table, filter, indexName);
+    return query(this.dbClient, this.table, filter, indexName, consistentRead);
   }
 
   async queryWithCursor<T extends AnyObject>(
     filter: Filter<T>,
     indexName?: string,
+    consistentRead?: boolean,
   ): Promise<{ items: Array<T>; cursor?: string; scannedCount: number }> {
-    return queryWithCursor(this.dbClient, this.table, filter, indexName);
+    return queryWithCursor(this.dbClient, this.table, filter, indexName, consistentRead);
   }
 
   async getItem<T extends AnyObject>(
     key: Key,
     fields?: Array<keyof T>,
+    consistentRead?: boolean,
   ): Promise<T> {
-    return getItem(this.dbClient, this.table, key, fields);
+    return getItem(this.dbClient, this.table, key, fields, consistentRead);
   }
 
   async batchGetItems(
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     keys: Array<{ [name: string]: any }>,
     fields?: Array<string>,
+    consistentRead?: boolean,
   ): Promise<Array<AnyObject>> {
-    return batchGetItems(this.dbClient, this.table, keys, fields);
+    return batchGetItems(this.dbClient, this.table, keys, fields, consistentRead);
   }
 
   async exists(key: Key): Promise<boolean> {

src/query/batchGetItems.spec.ts

@@ -167,4 +167,47 @@ describe('batchGetItems', () => {
       { pk: '5', sk: '6', id: 'c' },
     ]);
   });
+
+  test('consistent read set to true', async () => {
+    await batchGetItems([{ pk: 'xxxx', sk: 'yyyy' }], undefined, true);
+    expect(spy).toHaveBeenCalledWith(expect.objectContaining({
+      input: {
+        RequestItems: {
+          [testTableConf.name]: {
+            Keys: [{ pk: 'xxxx', sk: 'yyyy' }],
+            ConsistentRead: true,
+          },
+        },
+      }
+    }));
+  });
+
+  test('consistent read set to false', async () => {
+    await batchGetItems([{ pk: 'xxxx', sk: 'yyyy' }], undefined, false);
+    expect(spy).toHaveBeenCalledWith(expect.objectContaining({
+      input: {
+        RequestItems: {
+          [testTableConf.name]: {
+            Keys: [{ pk: 'xxxx', sk: 'yyyy' }],
+            ConsistentRead: false,
+          },
+        },
+      }
+    }));
+  });
+
+  test('consistent read not specified', async () => {
+    await batchGetItems([{ pk: 'xxxx', sk: 'yyyy' }], undefined, undefined);
+    expect(spy).toHaveBeenCalledWith(expect.objectContaining({
+      input: {
+        RequestItems: {
+          [testTableConf.name]: {
+            Keys: [{ pk: 'xxxx', sk: 'yyyy' }],
+          },
+        },
+      }
+    }));
+    // Ensure ConsistentRead is not present in the params
+    expect((spy.mock.calls[0][0] as any).input.RequestItems[testTableConf.name].ConsistentRead).toBeUndefined();
+  });
 });

src/query/batchGetItems.ts

@@ -7,13 +7,16 @@ import { AnyObject, TableConfig, Key } from '../types';
 /**
  * Get many items from the db matching the provided keys
  * @param keys array of key maps. eg: [{ pk: '1', sk: '2'}]
+ * @param fields optional fields to project
+ * @param consistentRead optional consistent read flag
  * @returns list of items
  */
 export async function batchGetItems(
   dbClient: DynamoDBDocumentClient,
   table: TableConfig,
   keys: Array<Key>,
   fields?: Array<string>,
+  consistentRead?: boolean,
 ): Promise<Array<AnyObject>> {
   let result: BatchGetCommandOutput;
   let unProcessedKeys = [];
@@ -22,7 +25,7 @@ export async function batchGetItems(
   // https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchGetItem.html
   if (keys.length > 100) {
     const results = await Promise.all(
-      chunk(keys, 100).map(x => batchGetItems(dbClient, table, x)),
+      chunk(keys, 100).map(x => batchGetItems(dbClient, table, x, fields, consistentRead)),
     );
     return flatten(results);
   }
@@ -51,6 +54,7 @@ export async function batchGetItems(
           ProjectionExpression: fieldsToProject
             ? fieldsToProject.join(',')
             : undefined,
+          ...(consistentRead !== undefined && { ConsistentRead: consistentRead }),
         },
       },
     }));

src/query/getItem.spec.ts

@@ -71,4 +71,53 @@ describe('getItem', () => {
       }
     }));
   });
+
+  test('consistent read set to true', async () => {
+    spy.mockResolvedValue({ Item: { id: 'xxxx' } });
+
+    await getItem({ pk: 'xxxx', sk: 'yyyy' }, undefined, true);
+    expect(testClient.send).toHaveBeenCalledWith(expect.objectContaining({
+      input: {
+        TableName: testTableConf.name,
+        Key: {
+          pk: 'xxxx',
+          sk: 'yyyy',
+        },
+        ConsistentRead: true,
+      }
+    }));
+  });
+
+  test('consistent read set to false', async () => {
+    spy.mockResolvedValue({ Item: { id: 'xxxx' } });
+
+    await getItem({ pk: 'xxxx', sk: 'yyyy' }, undefined, false);
+    expect(testClient.send).toHaveBeenCalledWith(expect.objectContaining({
+      input: {
+        TableName: testTableConf.name,
+        Key: {
+          pk: 'xxxx',
+          sk: 'yyyy',
+        },
+        ConsistentRead: false,
+      }
+    }));
+  });
+
+  test('consistent read not specified', async () => {
+    spy.mockResolvedValue({ Item: { id: 'xxxx' } });
+
+    await getItem({ pk: 'xxxx', sk: 'yyyy' }, undefined, undefined);
+    expect(testClient.send).toHaveBeenCalledWith(expect.objectContaining({
+      input: {
+        TableName: testTableConf.name,
+        Key: {
+          pk: 'xxxx',
+          sk: 'yyyy',
+        },
+      }
+    }));
+    // Ensure ConsistentRead is not present in the params
+    expect((testClient.send as jest.Mock).mock.calls[0][0].input.ConsistentRead).toBeUndefined();
+  });
 });

src/query/getItem.ts

@@ -13,6 +13,7 @@ export async function getItem<T extends AnyObject>(
   table: TableConfig,
   key: Key,
   fields?: Array<keyof T>,
+  consistentRead?: boolean,
 ): Promise<T> {
   const index = table.indexes.default;
 
@@ -35,6 +36,10 @@ export async function getItem<T extends AnyObject>(
     params.ProjectionExpression = fields.join(',');
   }
 
+  if (consistentRead !== undefined) {
+    params.ConsistentRead = consistentRead;
+  }
+
   const result = await dbClient.send(new GetCommand(params));
 
   return result && result.Item ? (result.Item as T) : null;

src/query/query.spec.ts

@@ -90,4 +90,56 @@ describe('query', () => {
     });
     expect(testClient.send).toHaveBeenCalledTimes(2);
   });
+
+  test('consistent read set to true', async () => {
+    testClient.send = jest.fn().mockResolvedValue({ Items: [] });
+
+    await query({
+      where: {
+        pk: 'xxxx',
+      },
+    }, undefined, true);
+
+    expect(testClient.send).toHaveBeenCalledWith(expect.objectContaining({
+      input: expect.objectContaining({
+        TableName: testTableConf.name,
+        ConsistentRead: true,
+      })
+    }));
+  });
+
+  test('consistent read set to false', async () => {
+    testClient.send = jest.fn().mockResolvedValue({ Items: [] });
+
+    await query({
+      where: {
+        pk: 'xxxx',
+      },
+    }, undefined, false);
+
+    expect(testClient.send).toHaveBeenCalledWith(expect.objectContaining({
+      input: expect.objectContaining({
+        TableName: testTableConf.name,
+        ConsistentRead: false,
+      })
+    }));
+  });
+
+  test('consistent read not specified', async () => {
+    testClient.send = jest.fn().mockResolvedValue({ Items: [] });
+
+    await query({
+      where: {
+        pk: 'xxxx',
+      },
+    }, undefined, undefined);
+
+    expect(testClient.send).toHaveBeenCalledWith(expect.objectContaining({
+      input: expect.objectContaining({
+        TableName: testTableConf.name,
+      })
+    }));
+    // Ensure ConsistentRead is not present in the params
+    expect((testClient.send as jest.Mock).mock.calls[0][0].input.ConsistentRead).toBeUndefined();
+  });
 });

src/query/query.ts

@@ -5,13 +5,16 @@ import { buildQueryTableParams } from './queryBuilder';
 /**
  * Queries DynamoDB and returns list of matching items
  * @param {Filter<T>} filter query filter
+ * @param {string} indexName optional index name
+ * @param {boolean} consistentRead optional consistent read flag
  * @returns {Array<T>} list of matching items
  */
 export async function query<T extends AnyObject>(
   dbClient: DynamoDBDocumentClient,
   table: TableConfig,
   filter: Filter<T>,
   indexName?: string,
+  consistentRead?: boolean,
 ): Promise<Array<T>> {
   const { partitionKeyName, sortKeyName } = table.indexes[
     indexName || 'default'
@@ -24,6 +27,10 @@ export async function query<T extends AnyObject>(
     params.IndexName = indexName;
   }
 
+  if (consistentRead !== undefined) {
+    params.ConsistentRead = consistentRead;
+  }
+
   let lastEvaluatedKey;
   let items: T[] = [];