DynamoDB Single-Table Design skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "DynamoDB Single-Table Design"
  short_description: "Every access pattern is one Query, checked by script"
  default_prompt: "$dynamodb-single-table Design a DynamoDB single-table schema for customers, orders and order items"

#!/usr/bin/env python3
"""Check a DynamoDB single-table design against its access patterns.

Reads a JSON spec describing the table keys, GSIs, sample items and access
patterns, then verifies that every pattern is served by a Query or GetItem on
an existing index, with a key shape that at least one item actually writes.

Usage:
    python3 check_access_patterns.py design.json
    python3 check_access_patterns.py --example > design.json

Exit code 0 when there are no errors (warnings allowed), 1 otherwise.
"""
from __future__ import annotations

import json
import sys

SK_OPS = {"eq", "begins_with", "between", "lt", "lte", "gt", "gte", "none"}

EXAMPLE = {
    "table": {"name": "app", "pk": "PK", "sk": "SK"},
    "indexes": [
        {"name": "GSI1", "pk": "GSI1PK", "sk": "GSI1SK", "projection": "ALL"}
    ],
    "items": [
        {"entity": "Customer", "PK": "CUSTOMER#<customerId>", "SK": "PROFILE",
         "GSI1PK": "EMAIL#<email>", "GSI1SK": "CUSTOMER"},
        {"entity": "Order", "PK": "CUSTOMER#<customerId>", "SK": "ORDER#<createdAt>#<orderId>",
         "GSI1PK": "ORDER#<orderId>", "GSI1SK": "ORDER"},
        {"entity": "OrderItem", "PK": "ORDER#<orderId>", "SK": "ITEM#<sku>"}
    ],
    "patterns": [
        {"name": "Get customer profile", "index": "table",
         "pk": "CUSTOMER#<customerId>", "sk": {"op": "eq", "value": "PROFILE"}},
        {"name": "Customer orders, newest first", "index": "table",
         "pk": "CUSTOMER#<customerId>", "sk": {"op": "begins_with", "value": "ORDER#"}},
        {"name": "Find customer by email", "index": "GSI1",
         "pk": "EMAIL#<email>", "sk": {"op": "eq", "value": "CUSTOMER"}},
        {"name": "Get order by id", "index": "GSI1",
         "pk": "ORDER#<orderId>", "sk": {"op": "eq", "value": "ORDER"}},
        {"name": "Items of an order", "index": "table",
         "pk": "ORDER#<orderId>", "sk": {"op": "begins_with", "value": "ITEM#"}}
    ]
}


def prefix(template: str) -> str:
    """Literal text before the first <placeholder>."""
    return template.split("<", 1)[0]


def shapes_match(pattern_value: str, item_value: str, op: str) -> bool:
    """True if an item key template could satisfy the pattern's key condition."""
    p, i = prefix(pattern_value), prefix(item_value)
    if op == "eq":
        if "<" not in pattern_value and "<" not in item_value:
            return pattern_value == item_value
        return i.startswith(p) or p.startswith(i)
    # begins_with, between and range operators: compare literal prefixes
    return i.startswith(p) or p.startswith(i)


def main(argv: list[str]) -> int:
    if len(argv) == 2 and argv[1] == "--example":
        print(json.dumps(EXAMPLE, indent=2))
        return 0
    if len(argv) != 2:
        print(__doc__.strip(), file=sys.stderr)
        return 2
    try:
        with open(argv[1], encoding="utf-8") as fh:
            spec = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        print(f"error: cannot read spec: {exc}", file=sys.stderr)
        return 2

    errors: list[str] = []
    warnings: list[str] = []
    table = spec.get("table") or {}
    if not table.get("pk"):
        errors.append("table.pk is required")
    indexes = {"table": {"name": "table", "pk": table.get("pk"), "sk": table.get("sk"), "base": True}}
    for idx in spec.get("indexes", []):
        if not idx.get("name") or not idx.get("pk"):
            errors.append(f"index {idx!r} needs name and pk")
            continue
        indexes[idx["name"]] = {**idx, "base": False}
    if len(indexes) - 1 > 20:
        warnings.append("more than 20 GSIs: above the default per-table quota")

    items = spec.get("items", [])
    for it in items:
        for attr in (table.get("pk"), table.get("sk")):
            if attr and attr not in it:
                errors.append(f"item '{it.get('entity', '?')}' is missing table key attribute {attr}")

    for name, idx in indexes.items():
        if idx["base"] or not items:
            continue
        if not any(idx["pk"] in it for it in items):
            warnings.append(f"index {name}: no sample item writes {idx['pk']}, so the index stays empty")

    rows = []
    for pat in spec.get("patterns", []):
        pname = pat.get("name", "?")
        iname = pat.get("index", "table")
        if pat.get("scan"):
            warnings.append(f"'{pname}': uses Scan; acceptable only for rare admin or export jobs")
            rows.append((pname, iname, "Scan", "WARN"))
            continue
        idx = indexes.get(iname)
        if idx is None:
            errors.append(f"'{pname}': index '{iname}' is not defined")
            rows.append((pname, iname, "-", "ERROR"))
            continue
        pk_t = pat.get("pk", "")
        sk = pat.get("sk") or {"op": "none"}
        op = sk.get("op", "none")
        status = "OK"
        if not pk_t:
            errors.append(f"'{pname}': pk template missing")
            status = "ERROR"
        if op not in SK_OPS:
            errors.append(f"'{pname}': unknown sort key operator '{op}'")
            status = "ERROR"
        if op != "none" and not idx.get("sk"):
            errors.append(f"'{pname}': index '{iname}' has no sort key, cannot apply '{op}'")
            status = "ERROR"
        if pat.get("consistent") and not idx["base"]:
            errors.append(f"'{pname}': strongly consistent reads are not supported on GSIs")
            status = "ERROR"
        if pk_t and "<" not in pk_t:
            warnings.append(f"'{pname}': constant partition key '{pk_t}' concentrates traffic on one partition")
            status = "WARN" if status == "OK" else status
        if pat.get("filter"):
            warnings.append(f"'{pname}': FilterExpression runs after the read; you pay for filtered-out items")
            status = "WARN" if status == "OK" else status
        if items and pk_t and status != "ERROR":
            candidates = [it for it in items
                          if idx["pk"] in it and shapes_match(pk_t, str(it[idx["pk"]]), "eq")]
            if not candidates:
                errors.append(f"'{pname}': no item writes a {idx['pk']} shaped like '{pk_t}'")
                status = "ERROR"
            elif op not in ("none",) and idx.get("sk"):
                val = str(sk.get("value", ""))
                if val and not any(idx["sk"] in it and shapes_match(val, str(it[idx["sk"]]), op)
                                   for it in candidates):
                    errors.append(f"'{pname}': no item under '{pk_t}' has {idx['sk']} matching {op} '{val}'")
                    status = "ERROR"
        kind = "GetItem" if (idx["base"] and op == "eq" and "<" not in str(sk.get("value", ""))) else "Query"
        rows.append((pname, iname, kind, status))

    if not rows:
        errors.append("no access patterns defined")

    width = max([len(r[0]) for r in rows] + [7])
    print(f"{'Pattern'.ljust(width)}  {'Index':<8} {'Op':<14} Status")
    for r in rows:
        print(f"{r[0].ljust(width)}  {r[1]:<8} {r[2]:<14} {r[3]}")
    for w in warnings:
        print(f"warn: {w}")
    for e in errors:
        print(f"error: {e}")
    print(f"\n{len(rows)} pattern(s), {len(errors)} error(s), {len(warnings)} warning(s)")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))