Added abort signal support for dynamo readonly functions

src/batch-get.ts

@@ -5,6 +5,7 @@ import { type Table } from './table'
 import { buildProjectionExpression } from './query/projection-expression'
 import { HelpfulError } from './errors'
 import { type AttributeMap } from './interfaces'
+import { type IRequestOptions } from './connections'
 
 export class BatchGet<T extends Table> {
   public static readonly MAX_BATCH_ITEMS = 100
@@ -69,7 +70,7 @@ export class BatchGet<T extends Table> {
     return this
   }
 
-  public async retrieve(): Promise<T[]> {
+  public async retrieve(requestOptions?: IRequestOptions): Promise<T[]> {
     const chunkSize = this.atomicity ? BatchGet.MAX_TRANSACT_ITEMS : BatchGet.MAX_BATCH_ITEMS
     return await Promise.all(
       chunk(this.items, chunkSize).map(async (chunkedItems) => {
@@ -117,11 +118,11 @@ export class BatchGet<T extends Table> {
           if (this.atomicity) {
             output = await this.dynamo.transactGetItems({
               TransactItems: transactItems,
-            })
+            }, requestOptions)
           } else {
             output = await this.dynamo.batchGetItem({
               RequestItems: requestMap,
-            })
+            }, requestOptions)
           }
         } catch (ex) {
           throw new HelpfulError(ex)
@@ -181,8 +182,8 @@ export class BatchGet<T extends Table> {
     })
   }
 
-  public async retrieveMapped(): Promise<Map<typeof Table, T[]>> {
-    const items = await this.retrieve()
+  public async retrieveMapped(requestOptions?: IRequestOptions): Promise<Map<typeof Table, T[]>> {
+    const items = await this.retrieve(requestOptions)
     const map = new Map<typeof Table, T[]>()
 
     for (const item of items) {

src/connections/index.ts

@@ -1,2 +1,3 @@
 export * from './connection'
 export * from './dynamodb-connection'
+export * from './request-options'

src/connections/request-options.ts

@@ -0,0 +1,7 @@
+export interface IRequestOptions {
+  abortSignal?: AbortSignal
+}
+
+export function isRequestOptions(value: unknown): value is IRequestOptions {
+  return typeof value === 'object' && value != null && 'abortSignal' in value
+}

src/query/global-secondary-index.ts

@@ -9,6 +9,7 @@ import { type Filters as QueryFilters } from './filters'
 import { QueryOutput } from './output'
 import { buildProjectionExpression } from './projection-expression'
 import { MagicSearch, type MagicSearchInput } from './search'
+import { type IRequestOptions } from '../connections'
 
 interface GlobalSecondaryIndexGetInput {
   /**
@@ -102,7 +103,7 @@ export class GlobalSecondaryIndex<T extends Table> {
    *
    * Avoid use whenever you do not have uniqueness for the GlobalSecondaryIndex's HASH + RANGE.
    */
-  public async get(filters: QueryFilters<T>, input: GlobalSecondaryIndexGetInput = {}): Promise<T | undefined> {
+  public async get(filters: QueryFilters<T>, input: GlobalSecondaryIndexGetInput = {}, requestOptions?: IRequestOptions): Promise<T | undefined> {
     if (!has(filters, this.metadata.hash.propertyName)) {
       throw new QueryError('Cannot perform .get() on a GlobalSecondaryIndex without specifying a hash key value')
     } else if (this.metadata.range != null && !has(filters, this.metadata.range.propertyName)) {
@@ -117,7 +118,7 @@ export class GlobalSecondaryIndex<T extends Table> {
     // DynamoDB will start the search at the first match and limit means it will only process
     // that document and return it, however, you cannot use any additional filters on this .get
     // method; for that, you need to use .query()
-    const results = await this.query(filters, input)
+    const results = await this.query(filters, input, requestOptions)
 
     if (results.count > 0) {
       return results[0]
@@ -177,7 +178,7 @@ export class GlobalSecondaryIndex<T extends Table> {
     return queryInput
   }
 
-  public async query(filters: QueryFilters<T>, input?: GlobalSecondaryIndexQueryInput): Promise<QueryOutput<T>> {
+  public async query(filters: QueryFilters<T>, input?: GlobalSecondaryIndexQueryInput, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     if (!has(filters, this.metadata.hash.propertyName)) {
       throw new QueryError('Cannot perform a query on a GlobalSecondaryIndex without specifying a hash key value')
     } else if (isArray(get(filters, this.metadata.hash.propertyName)) && get(filters, this.metadata.hash.propertyName)[0] !== '=') {
@@ -189,7 +190,7 @@ export class GlobalSecondaryIndex<T extends Table> {
     let output: QueryCommandOutput
 
     try {
-      output = await this.tableClass.schema.dynamo.query(queryInput)
+      output = await this.tableClass.schema.dynamo.query(queryInput, requestOptions)
     } catch (ex) {
       throw new HelpfulError(ex, this.tableClass, queryInput)
     }
@@ -241,12 +242,12 @@ export class GlobalSecondaryIndex<T extends Table> {
    * *WARNING*: In most circumstances this is not a good thing to do.
    * This will return all the items in this index, does not perform well!
    */
-  public async scan(filters?: QueryFilters<T> | undefined | null, input: GlobalSecondaryIndexScanInput = {}): Promise<QueryOutput<T>> {
+  public async scan(filters?: QueryFilters<T> | undefined | null, input: GlobalSecondaryIndexScanInput = {}, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const scanInput = this.getScanInput(input, filters == null ? undefined : filters)
     const hasProjection = scanInput.ProjectionExpression == null
     let output: ScanCommandOutput
     try {
-      output = await this.tableClass.schema.dynamo.scan(scanInput)
+      output = await this.tableClass.schema.dynamo.scan(scanInput, requestOptions)
     } catch (ex) {
       throw new HelpfulError(ex, this.tableClass, scanInput)
     }
@@ -264,11 +265,11 @@ export class GlobalSecondaryIndex<T extends Table> {
    *
    * @see {@link https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Scan.html#Scan.ParallelScan}
    */
-  public async segmentedScan(filters: QueryFilters<T> | undefined | null, input: GlobalSecondaryIndexSegmentedScanInput): Promise<QueryOutput<T>> {
+  public async segmentedScan(filters: QueryFilters<T> | undefined | null, input: GlobalSecondaryIndexSegmentedScanInput, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const scans: Array<Promise<QueryOutput<T>>> = []
     for (let i = 0; i < input.totalSegments; i++) {
       input.segment = i
-      scans.push(this.scan(filters, input))
+      scans.push(this.scan(filters, input, requestOptions))
     }
 
     const scanOutputs = await Promise.all(scans)

src/query/local-secondary-index.ts

@@ -8,6 +8,7 @@ import { buildQueryExpression } from './expression'
 import { type Filters as QueryFilters } from './filters'
 import { QueryOutput } from './output'
 import { MagicSearch, type MagicSearchInput } from './search'
+import { type IRequestOptions } from '../connections'
 
 interface LocalSecondaryIndexQueryInput {
   rangeOrder?: 'ASC' | 'DESC'
@@ -50,7 +51,7 @@ export class LocalSecondaryIndex<T extends Table> {
     return queryInput
   }
 
-  public async query(filters: QueryFilters<T>, input: LocalSecondaryIndexQueryInput = {}): Promise<QueryOutput<T>> {
+  public async query(filters: QueryFilters<T>, input: LocalSecondaryIndexQueryInput = {}, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     if (!has(filters, this.tableClass.schema.primaryKey.hash.propertyName)) {
       throw new QueryError('Cannot perform a query on a LocalSecondaryIndex without specifying a hash key value')
     }
@@ -70,7 +71,7 @@ export class LocalSecondaryIndex<T extends Table> {
     const hasProjection = queryInput.ProjectionExpression == null
     let output: QueryCommandOutput
     try {
-      output = await this.tableClass.schema.dynamo.query(queryInput)
+      output = await this.tableClass.schema.dynamo.query(queryInput, requestOptions)
     } catch (ex) {
       throw new HelpfulError(ex, this.tableClass, queryInput)
     }
@@ -91,7 +92,7 @@ export class LocalSecondaryIndex<T extends Table> {
     return scanInput
   }
 
-  public async scan(filters: QueryFilters<T> | undefined | null, input: LocalSecondaryIndexScanInput = {}): Promise<QueryOutput<T>> {
+  public async scan(filters: QueryFilters<T> | undefined | null, input: LocalSecondaryIndexScanInput = {}, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const scanInput = this.getScanInput(input)
     if (filters != null && Object.keys(filters).length > 0) {
       // don't pass the index metadata, avoids KeyConditionExpression
@@ -103,7 +104,7 @@ export class LocalSecondaryIndex<T extends Table> {
     const hasProjection = scanInput.ProjectionExpression == null
     let output: ScanCommandOutput
     try {
-      output = await this.tableClass.schema.dynamo.scan(scanInput)
+      output = await this.tableClass.schema.dynamo.scan(scanInput, requestOptions)
     } catch (ex) {
       throw new HelpfulError(ex, this.tableClass, scanInput)
     }

src/query/primary-key.ts

@@ -14,6 +14,7 @@ import { type Filters as QueryFilters, type UpdateConditions } from './filters'
 import { QueryOutput } from './output'
 import { buildProjectionExpression } from './projection-expression'
 import { MagicSearch, type MagicSearchInput } from './search'
+import { type IRequestOptions, isRequestOptions } from '../connections'
 
 type PrimaryKeyType = string | number | Date
 // eslint-disable-next-line @typescript-eslint/no-invalid-void-type
@@ -127,10 +128,10 @@ export class PrimaryKey<T extends Table, HashKeyType extends PrimaryKeyType, Ran
    *
    *   `.get(instanceOfTable)`
    */
-  public async get(filters: QueryFilters<T>, input?: PrimaryKeyGetInput): Promise<T | undefined>
-  public async get(hash: HashKeyType, range: RangeKeyType, input?: PrimaryKeyGetInput): Promise<T | undefined>
-  public async get(record: T, input?: PrimaryKeyGetInput): Promise<T | undefined>
-  public async get(hash: HashKeyType | T | QueryFilters<T>, range?: RangeKeyType | PrimaryKeyGetInput, input?: PrimaryKeyGetInput): Promise<T | undefined> {
+  public async get(filters: QueryFilters<T>, input?: PrimaryKeyGetInput, requestOptions?: IRequestOptions): Promise<T | undefined>
+  public async get(hash: HashKeyType, range: RangeKeyType, input?: PrimaryKeyGetInput, requestOptions?: IRequestOptions): Promise<T | undefined>
+  public async get(record: T, input?: PrimaryKeyGetInput, requestOptions?: IRequestOptions): Promise<T | undefined>
+  public async get(hash: HashKeyType | T | QueryFilters<T>, range?: RangeKeyType | PrimaryKeyGetInput, input?: PrimaryKeyGetInput | IRequestOptions, requestOptions?: IRequestOptions): Promise<T | undefined> {
     let record: T
 
     if (isDyngooseTableInstance(hash)) {
@@ -143,14 +144,15 @@ export class PrimaryKey<T extends Table, HashKeyType extends PrimaryKeyType, Ran
       throw new QueryError('PrimaryKey.get called with unknown arguments')
     }
 
-    const getGetInput: Partial<PrimaryKeyGetGetItemInput> = input == null ? ((range == null || isKeyValue(range)) ? {} : range as PrimaryKeyGetInput) : input
+    const options = isRequestOptions(requestOptions) ? requestOptions : isRequestOptions(input) ? input : undefined
+    const getGetInput: Partial<PrimaryKeyGetGetItemInput> = input == null ? ((range == null || isKeyValue(range)) ? {} : range as PrimaryKeyGetInput) : isRequestOptions(input) ? {} : input
     getGetInput.key = record.getDynamoKey()
     const getItemInput = this.getGetInput(getGetInput as PrimaryKeyGetGetItemInput)
     const hasProjection = getItemInput.ProjectionExpression == null
     let dynamoRecord: GetItemOutput
 
     try {
-      dynamoRecord = await this.table.schema.dynamo.getItem(getItemInput)
+      dynamoRecord = await this.table.schema.dynamo.getItem(getItemInput, options)
     } catch (ex) {
       throw new HelpfulError(ex, this.table, getItemInput)
     }
@@ -168,14 +170,14 @@ export class PrimaryKey<T extends Table, HashKeyType extends PrimaryKeyType, Ran
    *
    * @deprecated
    */
-  public async batchGet(inputs: Array<PrimaryKeyBatchInput<HashKeyType, RangeKeyType>>): Promise<T[]> {
+  public async batchGet(inputs: Array<PrimaryKeyBatchInput<HashKeyType, RangeKeyType>>, requestOptions?: IRequestOptions): Promise<T[]> {
     const batch = new BatchGet<T>()
 
     for (const input of inputs) {
       batch.get(this.fromKey(input[0], input[1]))
     }
 
-    return await batch.retrieve()
+    return await batch.retrieve(requestOptions)
   }
 
   /**
@@ -234,7 +236,7 @@ export class PrimaryKey<T extends Table, HashKeyType extends PrimaryKeyType, Ran
     return queryInput
   }
 
-  public async query(filters: QueryFilters<T>, input?: PrimaryKeyQueryInput): Promise<QueryOutput<T>> {
+  public async query(filters: QueryFilters<T>, input?: PrimaryKeyQueryInput, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     if (!has(filters, this.metadata.hash.propertyName)) {
       throw new QueryError('Cannot perform a query on the PrimaryKey index without specifying a hash key value')
     } else if (isArray(get(filters, this.metadata.hash.propertyName)) && get(filters, this.metadata.hash.propertyName)[0] !== '=') {
@@ -251,7 +253,7 @@ export class PrimaryKey<T extends Table, HashKeyType extends PrimaryKeyType, Ran
     let output: QueryCommandOutput
 
     try {
-      output = await this.table.schema.dynamo.query(queryInput)
+      output = await this.table.schema.dynamo.query(queryInput, requestOptions)
     } catch (ex) {
       throw new HelpfulError(ex, this.table, queryInput)
     }
@@ -294,11 +296,11 @@ export class PrimaryKey<T extends Table, HashKeyType extends PrimaryKeyType, Ran
     return scanInput
   }
 
-  public async scan(filters?: QueryFilters<T> | undefined | null, input?: PrimaryKeyScanInput): Promise<QueryOutput<T>> {
+  public async scan(filters?: QueryFilters<T> | undefined | null, input?: PrimaryKeyScanInput, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const scanInput = this.getScanInput(input, filters == null ? undefined : filters)
     let output: ScanCommandOutput
     try {
-      output = await this.table.schema.dynamo.scan(scanInput)
+      output = await this.table.schema.dynamo.scan(scanInput, requestOptions)
     } catch (ex) {
       throw new HelpfulError(ex, this.table, scanInput)
     }

src/query/search.ts

@@ -13,6 +13,7 @@ import { type LocalSecondaryIndex } from './local-secondary-index'
 import { QueryOutput } from './output'
 import { type PrimaryKey } from './primary-key'
 import { buildProjectionExpression } from './projection-expression'
+import { type IRequestOptions } from '../connections'
 
 type Index<T extends Table> = PrimaryKey<T, any, any> | GlobalSecondaryIndex<T> | LocalSecondaryIndex<T> | string
 
@@ -231,9 +232,9 @@ export class MagicSearch<T extends Table> {
    *
    * A promise will be returned that will resolve to the results array upon completion.
    */
-  async exec(): Promise<QueryOutput<T>> {
+  async exec(requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const input = this.getInput()
-    return await this.page(input)
+    return await this.page(input, requestOptions)
   }
 
   /**
@@ -249,7 +250,7 @@ export class MagicSearch<T extends Table> {
    * It is recommended you apply a `.limit(minOrMore)` before calling `.minimum` to ensure
    * you do not load too many results as well.
   */
-  async minimum(minimum: number): Promise<QueryOutput<T>> {
+  async minimum(minimum: number, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const input = this.getInput()
     const outputs: Array<QueryOutput<T>> = []
     let page: QueryOutput<T> | undefined
@@ -260,7 +261,7 @@ export class MagicSearch<T extends Table> {
         input.ExclusiveStartKey = page.lastEvaluatedKey
       }
 
-      page = await this.page(input)
+      page = await this.page(input, requestOptions)
       count += page.count
       outputs.push(page)
 
@@ -280,7 +281,7 @@ export class MagicSearch<T extends Table> {
    * This is also non-ideal for scans, for better performance use a segmented scan
    * via the Query.PrimaryKey.segmentedScan or Query.GlobalSecondaryIndex.segmentedScan.
    */
-  async all(): Promise<QueryOutput<T>> {
+  async all(requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const input = this.getInput()
     const outputs: Array<QueryOutput<T>> = []
     let page: QueryOutput<T> | undefined
@@ -291,7 +292,7 @@ export class MagicSearch<T extends Table> {
         input.ExclusiveStartKey = page.lastEvaluatedKey
       }
 
-      page = await this.page(input)
+      page = await this.page(input, requestOptions)
       outputs.push(page)
     }
 
@@ -404,14 +405,14 @@ export class MagicSearch<T extends Table> {
     return await this.exec()
   }
 
-  async page(input: ScanCommandInput | QueryCommandInput): Promise<QueryOutput<T>> {
+  async page(input: ScanCommandInput | QueryCommandInput, requestOptions?: IRequestOptions): Promise<QueryOutput<T>> {
     const hasProjection = input.ProjectionExpression == null
     let output: ScanCommandOutput | QueryCommandOutput
 
     // if we are filtering based on key conditions, run a query instead of a scan
     if ((input as QueryCommandInput).KeyConditionExpression != null) {
       try {
-        output = await this.tableClass.schema.dynamo.query(input)
+        output = await this.tableClass.schema.dynamo.query(input, requestOptions)
       } catch (ex) {
         throw new HelpfulError(ex, this.tableClass, input)
       }
@@ -423,7 +424,7 @@ export class MagicSearch<T extends Table> {
       }
 
       try {
-        output = await this.tableClass.schema.dynamo.scan(input)
+        output = await this.tableClass.schema.dynamo.scan(input, requestOptions)
       } catch (ex) {
         throw new HelpfulError(ex, this.tableClass, input)
       }

src/tables/describe-table.ts

@@ -1,11 +1,12 @@
 import { type DescribeTableCommandInput, type TableDescription } from '@aws-sdk/client-dynamodb'
 import { type Schema } from './schema'
+import { type IRequestOptions } from '../connections'
 
-export async function describeTable(schema: Schema): Promise<TableDescription> {
+export async function describeTable(schema: Schema, requestOptions?: IRequestOptions): Promise<TableDescription> {
   const params: DescribeTableCommandInput = {
     TableName: schema.name,
   }
 
-  const result = await schema.dynamo.describeTable(params)
+  const result = await schema.dynamo.describeTable(params, requestOptions)
   return result.Table as TableDescription
 }