From python
Designs TypeScript DynamoDB single-table schemas using dynamodb-toolbox v2: entities, PK/SK patterns, GSIs, queries, pagination.
npx claudepluginhub martinffx/atelier --plugin pythonThis skill uses the workspace's default tool permissions.
Type-safe DynamoDB interactions with Entity and Table abstractions for single-table design.
Provides TypeScript patterns for DynamoDB-Toolbox v2: schema/table/entity modeling, .build() workflows, query/scan access patterns, batch/transaction operations, single-table designs with computed keys. For type-safe DynamoDB access layers in TS services.
Guides AWS DynamoDB single-table design, GSI patterns, and SDK v3 usage in TypeScript/Python for access-pattern-first data modeling. Use for table schemas and queries.
Guides AWS DynamoDB table schemas, queries, indexes, capacity modes, single-table design, CRUD ops via CLI/boto3, and performance troubleshooting.
Share bugs, ideas, or general feedback.
Type-safe DynamoDB interactions with Entity and Table abstractions for single-table design.
✓ Use when:
✗ Avoid when:
DynamoDB inverts the relational paradigm: design for known access patterns, not flexible querying.
Before implementing:
See references/modeling.md for detailed methodology.
import { Table } from 'dynamodb-toolbox/table'
const AppTable = new Table({
name: process.env.TABLE_NAME || "AppTable",
partitionKey: { name: "PK", type: "string" },
sortKey: { name: "SK", type: "string" },
indexes: {
GSI1: {
type: "global",
partitionKey: { name: "GSI1PK", type: "string" },
sortKey: { name: "GSI1SK", type: "string" },
},
GSI2: {
type: "global",
partitionKey: { name: "GSI2PK", type: "string" },
sortKey: { name: "GSI2SK", type: "string" },
},
GSI3: {
type: "global",
partitionKey: { name: "GSI3PK", type: "string" },
sortKey: { name: "GSI3SK", type: "string" },
},
},
entityAttributeSavedAs: "_et", // default, customize if needed
});
| Index | Purpose |
|---|---|
| Main Table (PK/SK) | Primary entity access |
| GSI1 | Collection queries (issues by repo, members by org) |
| GSI2 | Entity-specific queries and relationships (forks) |
| GSI3 | Hierarchical queries with temporal sorting (repos by owner) |
import { Entity } from 'dynamodb-toolbox/entity'
import { item } from 'dynamodb-toolbox/schema/item'
import { string } from 'dynamodb-toolbox/schema/string'
const UserEntity = new Entity({
name: "USER",
table: AppTable,
schema: item({
// Business attributes
username: string().required().key(),
email: string().required(),
bio: string().optional(),
}).and(_schema => ({
// Computed keys (PK/SK/GSI keys derived from business attributes)
PK: string().key().link<typeof _schema>(
({ username }) => `ACCOUNT#${username}`
),
SK: string().key().link<typeof _schema>(
({ username }) => `ACCOUNT#${username}`
),
GSI1PK: string().link<typeof _schema>(
({ username }) => `ACCOUNT#${username}`
),
GSI1SK: string().link<typeof _schema>(
({ username }) => `ACCOUNT#${username}`
),
})),
});
const RepoEntity = new Entity({
name: "REPO",
table: AppTable,
schema: item({
owner: string()
.required()
.validate((value: string) => /^[a-zA-Z0-9_-]+$/.test(value))
.key(),
repo_name: string()
.required()
.validate((value: string) => /^[a-zA-Z0-9_-]+$/.test(value))
.key(),
description: string().optional(),
is_private: boolean().default(false),
}).and(_schema => ({
PK: string().key().link<typeof _schema>(
({ owner, repo_name }) => `REPO#${owner}#${repo_name}`
),
SK: string().key().link<typeof _schema>(
({ owner, repo_name }) => `REPO#${owner}#${repo_name}`
),
// GSI3 for temporal sorting (repos by owner, newest first)
GSI3PK: string().link<typeof _schema>(
({ owner }) => `ACCOUNT#${owner}`
),
GSI3SK: string()
.default(() => `#${new Date().toISOString()}`)
.savedAs("GSI3SK"),
})),
});
| Entity | PK | SK | Purpose |
|---|---|---|---|
| User | ACCOUNT#{username} | ACCOUNT#{username} | Direct access |
| Repository | REPO#{owner}#{name} | REPO#{owner}#{name} | Direct access |
| Issue | ISSUE#{owner}#{repo}#{padded_num} | Same as PK | Direct access + enumeration |
| Comment | REPO#{owner}#{repo} | ISSUE#{padded_num}#COMMENT#{id} | Comments under issue |
| Star | ACCOUNT#{username} | STAR#{owner}#{repo}#{timestamp} | Adjacency list pattern |
Key Pattern Rules:
ENTITY#{id} - Simple identifierPARENT#{id}#CHILD#{id} - HierarchyTYPE#{category}#{identifier} - Categorization#{timestamp} - Temporal sorting (# prefix ensures ordering)import { type InputItem, type FormattedItem } from 'dynamodb-toolbox/entity'
// Type exports
type UserRecord = typeof UserEntity
type UserInput = InputItem<typeof UserEntity> // For writes
type UserFormatted = FormattedItem<typeof UserEntity> // For reads
// Usage in entities
class User {
static fromRecord(record: UserFormatted): User { /* ... */ }
toRecord(): UserInput { /* ... */ }
}
See references/entity-layer.md for transformation patterns.
import { PutItemCommand, GetItemCommand, DeleteItemCommand } from 'dynamodb-toolbox'
class UserRepository {
constructor(private entity: UserRecord) {}
// CREATE with duplicate check
async create(user: User): Promise<User> {
try {
const result = await this.entity
.build(PutItemCommand)
.item(user.toRecord())
.options({
condition: { attr: "PK", exists: false }, // Prevent duplicates
})
.send()
return User.fromRecord(result.ToolboxItem)
} catch (error) {
if (error instanceof ConditionalCheckFailedException) {
throw new DuplicateEntityError("User", user.username)
}
throw error
}
}
// GET by key
async get(username: string): Promise<User | undefined> {
const result = await this.entity
.build(GetItemCommand)
.key({ username })
.send()
return result.Item ? User.fromRecord(result.Item) : undefined
}
// UPDATE with existence check
async update(user: User): Promise<User> {
try {
const result = await this.entity
.build(PutItemCommand)
.item(user.toRecord())
.options({
condition: { attr: "PK", exists: true }, // Must exist
})
.send()
return User.fromRecord(result.ToolboxItem)
} catch (error) {
if (error instanceof ConditionalCheckFailedException) {
throw new EntityNotFoundError("User", user.username)
}
throw error
}
}
// DELETE
async delete(username: string): Promise<void> {
await this.entity.build(DeleteItemCommand).key({ username }).send()
}
}
See references/error-handling.md for error patterns.
import { QueryCommand } from 'dynamodb-toolbox/table/actions/query'
// List issues for a repository using GSI1
async listIssues(owner: string, repoName: string): Promise<Issue[]> {
const result = await this.table
.build(QueryCommand)
.entities(this.issueEntity)
.query({
partition: `ISSUE#${owner}#${repoName}`,
index: "GSI1",
})
.send()
return result.Items?.map(item => Issue.fromRecord(item)) || []
}
// List by status using beginsWith on SK
async listOpenIssues(owner: string, repoName: string): Promise<Issue[]> {
const result = await this.table
.build(QueryCommand)
.entities(this.issueEntity)
.query({
partition: `ISSUE#${owner}#${repoName}`,
index: "GSI4",
range: {
beginsWith: "ISSUE#OPEN#", // Filter to open issues only
},
})
.send()
return result.Items?.map(item => Issue.fromRecord(item)) || []
}
// Encode/decode pagination tokens
function encodePageToken(lastEvaluated?: Record<string, unknown>): string | undefined {
return lastEvaluated
? Buffer.from(JSON.stringify(lastEvaluated)).toString("base64")
: undefined
}
function decodePageToken(token?: string): Record<string, unknown> | undefined {
return token ? JSON.parse(Buffer.from(token, "base64").toString()) : undefined
}
// Query with pagination
async listReposByOwner(owner: string, limit = 50, offset?: string) {
const result = await this.table
.build(QueryCommand)
.entities(this.repoEntity)
.query({
partition: `ACCOUNT#${owner}`,
index: "GSI3",
range: { lt: "ACCOUNT#" }, // Filter to only repos (not account itself)
})
.options({
reverse: true, // Newest first
exclusiveStartKey: decodePageToken(offset), // Continue from cursor
limit,
})
.send()
return {
items: result.Items?.map(item => Repo.fromRecord(item)) || [],
nextOffset: encodePageToken(result.LastEvaluatedKey),
}
}
See references/transactions.md for:
$add(1)See references/testing.md for:
Schema:
item({}) for schema definition.key().and().link<typeof _schema>() to compute PK/SK/GSI keys.validate() for field validation.savedAs() when DynamoDB name differs from schema nameTypes:
InputItem<T> for writes (excludes computed attributes)FormattedItem<T> for reads (includes all attributes)Repository:
PutItemCommand with { attr: "PK", exists: false } for createsPutItemCommand with { attr: "PK", exists: true } for updatesGetItemCommand with .key() for readsQueryCommand with .entities() for type-safe queriesErrors:
ConditionalCheckFailedException → DuplicateEntityError (create) or EntityNotFoundError (update)Testing: