graphddb-runtime 0.2.2

Thin DynamoDB executor for GraphDDB-generated Python repositories (single-operation core, issue #44).

graphddb-runtime

Thin DynamoDB executor for GraphDDB-generated Python repositories.

graphddb-runtime is the small, hand-written package that generated repositories.py modules import as from graphddb_runtime import GraphDDBRuntime. It interprets the manifest.json / operations.json specifications produced by graphddb generate python and executes the validated access patterns against DynamoDB through boto3 — no scans, no hand-written key logic.

Features

• Single-operation core — GetItem / Query reads and PutItem / UpdateItem / DeleteItem writes.
• Relations & assembly — relation traversal, multi-operation assembly, BatchGetItem, result limits, and explain.
• Conditional & transactional writes — conditional writes, declarative transactions (execute_transaction, with forEach / when expansion and TransactWriteItems batching up to 25 items).
• Async adapter — AsyncGraphDDBRuntime exposes an await-able surface with behavior identical to the synchronous runtime.

Install

pip install graphddb-runtime

Requires Python 3.9+ and boto3.

Versioning. graphddb-runtime tracks the graphddb npm package version: a given runtime release matches the graphddb CLI of the same version, so the generated manifest.json / operations.json and the runtime that interprets them always stay in sync. Install the graphddb-runtime whose version equals the graphddb CLI you generated with.

Usage

Point the runtime at the two JSON specs emitted by graphddb generate python and pass a boto3 DynamoDB client. The generated repositories wrap it with typed methods:

import boto3
from graphddb_runtime import GraphDDBRuntime
from generated import UserRepository

runtime = GraphDDBRuntime(
    dynamodb_client=boto3.client("dynamodb"),
    manifest_path="generated/manifest.json",
    operations_path="generated/operations.json",
    # Map logical table names to deployed physical names when they differ.
    table_mapping={"UserPermissions": "UserPermissions-prod"},
)

users = UserRepository(runtime)
user = users.get_user_by_email(email="[e-mail]")

Async

boto3 is a synchronous SDK, so the runtime core is synchronous. AsyncGraphDDBRuntime is a thin adapter that runs each blocking call in a worker thread via asyncio.to_thread, giving an await-able surface with identical behavior (same params, specs, results, and error types). It does not require aioboto3.

import boto3
from graphddb_runtime import GraphDDBRuntime, AsyncGraphDDBRuntime

sync = GraphDDBRuntime(
    dynamodb_client=boto3.client("dynamodb"),
    manifest_path="generated/manifest.json",
    operations_path="generated/operations.json",
)
runtime = AsyncGraphDDBRuntime(sync)

user = await runtime.execute_query("getUser", {"userId": "alice"})
await runtime.execute_transaction("addManyMembers", {"groupId": "eng", "members": [...]})

The wrapped synchronous runtime is available as runtime.sync for callers that need the blocking API directly.

AWS Lambda

The runtime loads the JSON specs from disk and constructs a boto3 client — both are cold-start costs you want to pay once, in module scope, so they are reused across warm invocations (and frozen by SnapStart).

# handler.py — module scope runs once per execution environment (cold start).
import json
import boto3
from graphddb_runtime import GraphDDBRuntime
from generated import UserRepository

_runtime = GraphDDBRuntime(
    dynamodb_client=boto3.client("dynamodb"),
    manifest_path="generated/manifest.json",
    operations_path="generated/operations.json",
    table_mapping={"UserPermissions": "UserPermissions-prod"},
)
_users = UserRepository(_runtime)


def handler(event, context):
    user = _users.get_user_by_email(email=event["queryStringParameters"]["email"])
    if user is None:
        return {"statusCode": 404, "body": "not found"}
    return {"statusCode": 200, "body": json.dumps(user)}

SnapStart

Lambda SnapStart snapshots the initialized execution environment after the module-scope code runs, so the global client + GraphDDBRuntime(...) construction is captured in the snapshot and skipped on restore.

• Initialize the runtime and repositories in module scope (as above), never inside the handler — that is what gets snapshotted.
• Do not cache short-lived state across the snapshot (credentials/tokens with an expiry, random seeds). The DynamoDB client and the loaded specs are safe to snapshot; refresh anything time-sensitive inside the handler.

Packaging

The deployment artifact needs three things: this runtime package, the generated bindings, and the two JSON specs. boto3/botocore are provided by the Lambda Python runtime, so they need not be vendored (pin them only if you require a specific version).

mkdir -p build
pip install graphddb-runtime --target build    # the runtime
cp -r generated build/generated                # manifest.json, operations.json, *.py
cp handler.py build/
( cd build && zip -r ../function.zip . )        # handler = handler.handler

Make sure the manifest_path / operations_path you pass to GraphDDBRuntime resolve relative to the deployed working directory (e.g. generated/... when the specs are zipped under a generated/ folder at the artifact root).

License

MIT