@chainsaws/dynamodb

Type-safe DynamoDB wrapper for Node.js with a Python-style API, model-based type inference, partition metadata, and secondary-index helpers.

Requirements

• Node.js >= 22
• AWS credentials available through the normal AWS SDK resolution chain, or a local DynamoDB endpoint

Installation

npm install @chainsaws/dynamodb
yarn add @chainsaws/dynamodb
pnpm add @chainsaws/dynamodb

This package is ESM-only.

Quick Start

import {
  DynamoDBAPI,
  DynamoModel,
  defineModels,
} from "@chainsaws/dynamodb";

class User extends DynamoModel {
  static _partition = "user" as const;
  static _pk = "user_id" as const;
  static _sk = "created_at" as const;
  static _indexes = [{ pk: "email", sk: "user_id" }] as const;

  declare user_id: `u_${string}`;
  declare created_at: number;
  declare name: string;
  declare email: string | null;
  declare status: "active" | "inactive";
}

class Post extends DynamoModel {
  static _partition = "post" as const;
  static _pk = "post_id" as const;
  static _sk = "created_at" as const;

  declare post_id: string;
  declare created_at: number;
  declare author_id: string;
  declare title: string;
}

const models = defineModels({ User, Post });

const db = DynamoDBAPI.createDBFromModels(
  "my-table",
  {
    region: "ap-northeast-2",
    endpoint: process.env.DYNAMODB_ENDPOINT,
    max_pool_connections: 100,
  },
  models
);

await db.init_db_table();
await db.apply_model_partition(User, Post);

const created = await db.put_item("user", {
  user_id: "u_001",
  created_at: Date.now(),
  name: "Joon",
  email: "[email protected]",
  status: "active",
});

const loaded = await db.get_item(created._id);
console.log(loaded);

Package Surface

The package exports four main groups:

• DynamoDBAPI: high-level table client
• LowLevelDynamo: thin AWS SDK adapter
• DynamoModel, defineModel, defineModels: model-definition helpers
• error, filter, and type-inference helpers

Most applications should use DynamoDBAPI with model classes and only drop to LowLevelDynamo for explicitly AWS-shaped operations.

Concepts

Table initialization vs model initialization

There are two separate setup steps:

1. init_db_table() Creates the physical DynamoDB table if it does not exist and enables TTL on _ttl.

2. apply_model_partition(...models) or apply_partition_map(map) Writes partition metadata into the table and creates the GSIs declared by your models.

createDBFromModels(...) only gives you TypeScript inference. It does not create partitions by itself.

Item IDs

This package exposes item IDs as a merged string:

"<encoded _pk>&<encoded _sk>"

You usually do not need to build this manually. Read _id from a returned item and pass it back into get_item, update_item, or delete_item.

Model classes

To use model-based inference:

• Extend DynamoModel
• Define static _partition
• Define static _pk
• Define static _sk
• Optionally define static _indexes
• Add TypeScript instance fields for your item attributes
• Prefer declare fields so model declarations do not emit runtime undefined properties

Example:

class User extends DynamoModel {
  static _partition = "user" as const;
  static _pk = "user_id" as const;
  static _sk = "created_at" as const;
  static _indexes = [
    { pk: "email", sk: "user_id" },
    { pk: "_ptn", sk: "_crt" },
  ] as const;

  declare user_id: string;
  declare created_at: number;
  declare name: string;
  declare email: string | null;
}

Notes:

• _partition is the logical partition name stored in metadata.
• _pk and _sk are your logical key field names, not the physical DynamoDB attribute names.
• _indexes declares additional queryable key pairs.
• The library stores physical keys internally as _pk, _sk, _pk1, _sk1, etc.
• createDBFromModels(...), apply_model_partition(...), and defineModels(...) validate that _pk, _sk, and _indexes point to real model fields.

Model validation helpers

If you want an explicit checked registry, use defineModels(...):

import { defineModels } from "@chainsaws/dynamodb";

export const models = defineModels({ User, Post });

If you want the check to happen at model declaration time, use defineModel(...):

import { DynamoModel, defineModel } from "@chainsaws/dynamodb";

export const User = defineModel(
  class User extends DynamoModel {
    static _partition = "user" as const;
    static _pk = "user_id" as const;
    static _sk = "created_at" as const;
    static _indexes = [{ pk: "email", sk: "user_id" }] as const;

    declare user_id: string;
    declare created_at: number;
    declare email: string | null;
  }
);

Creating the Client

Typed client with model inference

const db = DynamoDBAPI.createDBFromModels(
  "my-table",
  { region: "us-east-1" },
  { User, Post }
);

This is the recommended path when you want autocomplete for:

• partition names
• put_item payloads
• query_items key-field pairs

Untyped client

const db = new DynamoDBAPI("my-table", {
  region: "us-east-1",
  endpoint: "http://localhost:8000",
});

Use this when you do not want model classes, or when partition definitions come from somewhere else.

Config

type DynamoDBAPIConfig = {
  region?: string;
  endpoint?: string;
  credentials?: {
    accessKeyId: string;
    secretAccessKey: string;
    sessionToken?: string;
  };
  max_pool_connections?: number;
};

• region: optional AWS region
• endpoint: optional custom endpoint, useful for DynamoDB Local
• credentials: optional static AWS credentials
• max_pool_connections: optional HTTP connection pool size for the AWS SDK client

When region or credentials are omitted, the AWS SDK uses its normal default resolution chain.

The shared config keys intentionally match the rest of the AWS packages in this repo: region, endpoint, credentials, and max_pool_connections.

Initializing Models and Partitions

From model classes

await db.init_db_table();
await db.apply_model_partition(User, Post);

This will:

• create or update partition metadata for each model
• create the default _ptn / _crt query index
• create custom indexes from each model's static _indexes

From a manual partition map

await db.apply_partition_map({
  user: {
    pk: "user_id",
    sk: "created_at",
    indexes: [
      { pk: "_ptn", sk: "_crt" },
      { pk: "email", sk: "user_id" },
    ],
  },
  post: {
    pk: "post_id",
    sk: "created_at",
  },
});

Use this when you do not want to define classes, or when partitions are configured dynamically.

CRUD

Create: put_item

With a plain object

const user = await db.put_item("user", {
  user_id: "u_001",
  created_at: Date.now(),
  name: "Joon",
  email: "[email protected]",
  status: "active",
});

With a model instance

const userModel = User.from_dict({
  user_id: "u_002",
  created_at: Date.now(),
  name: "Min",
  email: null,
  status: "inactive",
});

const savedModel = await db.put_item("user", userModel);

Behavior:

• required partition key and sort key fields must be present
• _id, _ptn, and _crt are added automatically
• if you pass a DynamoModel, the returned value is also a DynamoModel
• if you pass a plain object, the returned value is a plain object

Overwrite control:

await db.put_item("user", payload, false);

Passing false makes the write conditional on the item not already existing.

Read one: get_item

const item = await db.get_item(user._id);

if (!item) {
  console.log("not found");
} else {
  console.log(item.name, item._ptn, item._crt);
}

Behavior:

• returns null when the item does not exist
• returns a plain object, not a model instance
• strips internal _pk / _sk / GSI key attributes from the response

Read many: get_items

const items = await db.get_items([
  firstUserId,
  secondUserId,
  missingUserId,
]);

Behavior:

• preserves the same order as the input IDs
• missing items are returned as null

Example:

const [first, second, third] = await db.get_items(ids);
console.log(first?._id, second?._id, third); // third can be null

Update one: update_item

const updated = await db.update_item("user", user._id, {
  name: "Updated Name",
  status: "inactive",
});

Behavior:

• updates only the provided fields
• internal fields _pk, _sk, _id, _crt, _ptn are ignored from the payload
• throws if the item does not exist

Update many: update_items

const updatedItems = await db.update_items("user", {
  [firstUserId]: { status: "active" },
  [secondUserId]: { status: "inactive", name: "Renamed" },
});

Behavior:

• accepts a map of item_id -> partial update
• returns updated items in the same order as Object.entries(item_updates)
• currently runs sequentially in the Node port

Delete one: delete_item

const deleted = await db.delete_item(user._id);
console.log(deleted);

Behavior:

• throws if the item does not exist
• returns the deleted attributes as stored just before deletion
• this return value is useful for audit/logging/debugging

Delete many: delete_items

await db.delete_items([firstUserId, secondUserId]);

Behavior:

• deletes in DynamoDB batch-write chunks
• does not return deleted records

Create many: put_items

const createdUsers = await db.put_items("user", [
  {
    user_id: "u_101",
    created_at: Date.now(),
    name: "A",
    email: "[email protected]",
    status: "active",
  },
  {
    user_id: "u_102",
    created_at: Date.now() + 1,
    name: "B",
    email: "[email protected]",
    status: "inactive",
  },
]);

Behavior:

• chunks writes to DynamoDB limits internally
• supports can_overwrite = false just like put_item
• returns model instances if the input array contains model instances

Querying

query_items returns:

Promise<[items: ProcessedItem[], nextKey: string | null]>

Query by partition default order

const [latestUsers, nextKey] = await db.query_items("user");

If you omit pk_field and pk_value, the library queries the partition's default _ptn / _crt index.

Query by primary key fields

const [userItems] = await db.query_items("user", {
  pk_field: "user_id",
  pk_value: "u_001",
  sk_field: "created_at",
  sk_condition: "gte",
  sk_value: 0,
});

Query by a declared secondary index

If your model declares:

static _indexes = [{ pk: "email", sk: "user_id" }] as const;

then you can query:

const [byEmail] = await db.query_items("user", {
  pk_field: "email",
  pk_value: "[email protected]",
  sk_field: "user_id",
  sk_condition: "eq",
  sk_value: "u_001",
});

If the pk_field / sk_field pair does not match the primary key pair or a declared index pair, the library throws DynamoDBError.

Pagination

const [page1, nextKey] = await db.query_items("user", { limit: 50 });

if (nextKey) {
  const [page2] = await db.query_items("user", {
    limit: 50,
    start_key: nextKey,
  });
}

start_key must be the opaque cursor string returned by the previous call.

Reverse order

const [latestFirst] = await db.query_items("user", {
  reverse: true,
});

Consistent reads

const [items] = await db.query_items("user", {
  consistent_read: true,
});

When querying a GSI, DynamoDB does not support strongly consistent reads, so the library automatically uses false.

Simple filters

const [activeUsers] = await db.query_items("user", {
  filters: [
    { field: "status", condition: "eq", value: "active" },
    { field: "name", condition: "stw", value: "Jo" },
  ],
});

Multiple filters are combined with AND.

Recursive filters

const [filtered] = await db.query_items("user", {
  recursive_filters: {
    left: { field: "status", condition: "eq", value: "active" },
    operation: "and",
    right: {
      left: { field: "name", condition: "stw", value: "Jo" },
      operation: "or",
      right: { field: "email", condition: "exist" },
    },
  },
});

Simple filters and recursive_filters are combined together with AND.

Projection

const [slimUsers] = await db.query_items("user", {
  projection_fields: ["user_id", "email", "status"],
});

Behavior:

• only the requested fields are read from DynamoDB
• _id, _ptn, and _crt are still returned for consistency
• omitting projection_fields or passing null reads the full stored item instead

Streaming helpers

Use the iterator helpers when you want automatic pagination without managing nextKey manually.

for await (const item of db.iter_query("user", {
  pk_field: "user_id",
  pk_value: "u_001",
  limit: 100,
})) {
  console.log(item._id);
}

generate_items(...) is the Python-style alias for iter_query(...).

For full-table scans:

for await (const item of db.scan_items({ limit: 100 })) {
  console.log(item._ptn, item._id);
}

iter_scan(...) is the alias form of scan_items(...).

Behavior:

• scan iterators skip partition metadata rows automatically
• yielded rows use the same processed shape as query_items
• limit is treated as the per-page request size for each internal scan/query loop

Supported filter conditions

• eq
• neq
• lt
• lte
• gt
• gte
• btw
• not_btw
• stw
• not_stw
• is_in
• is_not_in
• contains
• not_contains
• exist
• not_exist

Field paths may use dotted access such as profile.nickname and indexed segments such as tags[0].

Partition Metadata Management

These helpers are useful when you need to manage partitions manually:

await db.create_partition("audit_log", "log_id", "created_at");
await db.append_index("audit_log", "actor_id", "created_at");

const names = await db.get_partition_names();

await db.update_partition("audit_log", "log_id", "created_at");
await db.detach_index("audit_log", "_pk2-_sk2");
await db.delete_partition("audit_log");

Notes:

• delete_partition(partition) removes all items in that logical partition and then deletes the partition metadata
• append_index creates both metadata and the physical GSI
• detach_index removes metadata and also attempts to delete the physical GSI

Return Types Summary

• put_item(partition, plainObject) -> plain object with _id, _ptn, _crt
• put_item(partition, modelInstance) -> model instance with _id, _ptn, _crt
• put_items(...) -> array of the same input style
• get_item(id) -> plain object or null
• get_items(ids) -> array of plain object | null
• update_item(...) -> plain object or model instance matching input style
• update_items(...) -> array of updated items
• delete_item(id) -> deleted attributes
• delete_items(ids) -> void
• query_items(...) -> [items, nextKey]
• iter_query(...) -> AsyncGenerator<ProcessedItem>
• generate_items(...) -> AsyncGenerator<ProcessedItem>
• scan_items(...) -> AsyncGenerator<ProcessedItem>
• iter_scan(...) -> AsyncGenerator<ProcessedItem>

Errors

The package exports:

• DynamoDBError
• DynamoDBPartitionError
• PartitionNotFoundError
• DuplicateIndexError
• BatchOperationError

Typical cases:

• missing partition metadata
• invalid query key pair
• conditional put failure when can_overwrite = false
• trying to delete or update a missing item
• batch failures

Development

pnpm --filter @chainsaws/dynamodb check-types
pnpm --filter @chainsaws/dynamodb test
pnpm --filter @chainsaws/dynamodb lint
pnpm --filter @chainsaws/dynamodb build

License

MIT