Skip to main content

dkmio

The most efficient way to interact with AWS DynamoDB in Python.

dkmio is an OKM (Object-Key Mapping) layer that provides a fluent API for building complex queries and operations. It handles expression building, attribute escaping, pagination, and error mapping automatically, allowing you to focus on your data instead of the low-level details of the DynamoDB API.

Core Philosophies​

  • Minimal Definition: Define only what's strictly necessary (PK, SK, and Indexes). Your attributes are free-form, embracing the NoSQL nature of DynamoDB.
  • Explicit over Implicit: A get() is always a GetItem, a query() is always a Query, and a scan() is always a Scan. No magic scans that consume your throughput silently.
  • Developer Productivity: Fluent API, auto-pagination, and type safety are baked in.

Top Features​

  • Fluent API: Chain methods naturally: .query().where().filter().select().limit().
  • Automatic Escaping: All attribute names are escaped to avoid DynamoDB's 700+ reserved words.
  • Auto-Execute: No need to call .execute(). Builders trigger execution on iteration, indexing, or len().
  • Auto-Pagination: .fetch_all() and .count() handle multi-page results seamlessly.
  • Batch Optimization: Auto-chunks and retries with exponential backoff for batch operations.
  • Smart Projections: Validates requested attributes against GSI/LSI projections before execution.
  • ACID Transactions: High-level API for transaction.write() and transaction.read().
  • Type Safety: Ships with py.typed marker for full mypy and pyright support.

Installation​

pip install dkmio

Required: boto3 >= 1.26.0

Quick Peek​

from dkmio import DynamoDB, PK, SK

db = DynamoDB(region_name="us-east-1")

class Orders(db.Table):
    __table_name__ = "orders"
    pk = PK("user_id")
    sk = SK("order_id")

orders = Orders()

# Get a single item
order = orders.get(user_id="usr_123", order_id="ord_456")

# Query with conditions — auto-executes on iteration
for order in orders.query(user_id="usr_123").where(gte="ord_100").filter(status__eq="NEW"):
    print(order["total"])

Next Steps​

Explore the documentation to learn more: