Skip to content

Indexes

DynamoDB supports two types of secondary indexes:

  • Global secondary index (GSI) - Query by any attribute with a different hash key
  • Local secondary index (LSI) - Query by the same hash key with a different sort key

Global secondary indexes

GSIs let you query by attributes other than the table's primary key. Define them as class attributes on your Model.

Key features

  • Query by any attribute, not just the primary key
  • Single or multi-attribute composite keys (up to 4 per key)
  • Range key conditions for efficient filtering
  • Automatic pagination with metrics

Define a GSI

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import GlobalSecondaryIndex


class User(Model):
    model_config = ModelConfig(table="users")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    email = StringAttribute()
    status = StringAttribute()
    age = NumberAttribute()

    # GSI with hash key only
    email_index = GlobalSecondaryIndex(
        index_name="email-index",
        partition_key="email",
    )

    # GSI with hash and range key
    status_index = GlobalSecondaryIndex(
        index_name="status-index",
        partition_key="status",
        sort_key="pk",
    )

Query a GSI

Use the index attribute to query. Pass the partition key value using partition_key=:

"""Query a GSI example."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import StringAttribute
from pydynox.indexes import GlobalSecondaryIndex

client = get_default_client()


class User(Model):
    model_config = ModelConfig(table="users_gsi")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    email = StringAttribute()
    status = StringAttribute()
    name = StringAttribute()

    # Define GSIs
    email_index = GlobalSecondaryIndex("email-index", partition_key="email")
    status_index = GlobalSecondaryIndex("status-index", partition_key="status")


async def main():
    # Setup: create table with GSIs
    if not await client.table_exists("users_gsi"):
        await client.create_table(
            "users_gsi",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            global_secondary_indexes=[
                {"index_name": "email-index", "hash_key": ("email", "S")},
                {"index_name": "status-index", "hash_key": ("status", "S")},
            ],
        )

    # Create some users
    await User(
        pk="USER#1", sk="PROFILE", email="john@example.com", status="active", name="John"
    ).save()
    await User(
        pk="USER#2", sk="PROFILE", email="jane@example.com", status="active", name="Jane"
    ).save()

    # Query by email (using partition_key=)
    async for user in User.email_index.query(partition_key="john@example.com"):
        print(f"By email: {user.name}")

    # Query by status (using partition_key=)
    async for user in User.status_index.query(partition_key="active"):
        print(f"Active: {user.name}")


asyncio.run(main())

Tip

All index queries use partition_key= as the first argument. This is the same pattern as Model.query(partition_key=...), so the API is consistent everywhere. You can also pass the attribute name as a keyword argument (e.g., email="john@example.com"), but partition_key= is the recommended way.

Range key conditions

When your GSI has a range key, you can add conditions:

"""GSI query with range key condition example."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import StringAttribute
from pydynox.indexes import GlobalSecondaryIndex

client = get_default_client()


class User(Model):
    model_config = ModelConfig(table="users_gsi_range")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    status = StringAttribute()
    name = StringAttribute()

    # GSI with status as hash key and pk as range key
    status_index = GlobalSecondaryIndex("status-index", partition_key="status", sort_key="pk")


async def main():
    # Setup: create table with GSI
    if not await client.table_exists("users_gsi_range"):
        await client.create_table(
            "users_gsi_range",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            global_secondary_indexes=[
                {
                    "index_name": "status-index",
                    "hash_key": ("status", "S"),
                    "range_key": ("pk", "S"),
                },
            ],
        )

    # Create users
    await User(pk="USER#001", sk="PROFILE", status="active", name="John").save()
    await User(pk="USER#050", sk="PROFILE", status="active", name="Jane").save()
    await User(pk="USER#100", sk="PROFILE", status="active", name="Bob").save()
    await User(pk="ADMIN#001", sk="PROFILE", status="active", name="Admin").save()

    # Query active users with pk starting with "USER#"
    print("Active users (pk starts with USER#):")
    async for user in User.status_index.query(
        partition_key="active",
        sort_key_condition=User.pk.begins_with("USER#"),
    ):
        print(f"  {user.name} ({user.pk})")

    # Query with comparison
    print("\nActive users (pk >= USER#050):")
    async for user in User.status_index.query(
        partition_key="active",
        sort_key_condition=User.pk >= "USER#050",
    ):
        print(f"  {user.name} ({user.pk})")


asyncio.run(main())

Filter conditions

Filter non-key attributes after the query:

"""GSI query with filter condition example."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import GlobalSecondaryIndex

client = get_default_client()


class User(Model):
    model_config = ModelConfig(table="users_gsi_filter")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    status = StringAttribute()
    name = StringAttribute()
    age = NumberAttribute()

    status_index = GlobalSecondaryIndex("status-index", partition_key="status")


async def main():
    # Setup: create table with GSI
    if not await client.table_exists("users_gsi_filter"):
        await client.create_table(
            "users_gsi_filter",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            global_secondary_indexes=[
                {"index_name": "status-index", "hash_key": ("status", "S")},
            ],
        )

    # Create users with different ages
    await User(pk="USER#1", sk="PROFILE", status="active", name="John", age=30).save()
    await User(pk="USER#2", sk="PROFILE", status="active", name="Jane", age=25).save()
    await User(pk="USER#3", sk="PROFILE", status="active", name="Bob", age=35).save()
    await User(pk="USER#4", sk="PROFILE", status="inactive", name="Alice", age=40).save()

    # Query active users over 30
    print("Active users age >= 30:")
    async for user in User.status_index.query(
        partition_key="active",
        filter_condition=User.age >= 30,
    ):
        print(f"  {user.name} (age={user.age})")


asyncio.run(main())

Warning

Filters run after the query. You still pay for RCU on filtered items.

Sort order

Control the sort order with scan_index_forward:

# Ascending (default)
async for user in User.status_index.query(partition_key="active", scan_index_forward=True):
    print(user.name)

# Descending
async for user in User.status_index.query(partition_key="active", scan_index_forward=False):
    print(user.name)

Pagination

Understanding limit vs page_size

GSI queries support the same pagination parameters as table queries:

Parameter What it does DynamoDB behavior
limit Max total items to return Stops iteration after N items
page_size Items per DynamoDB request Passed as Limit to DynamoDB API

Key behaviors:

  • limit=10 → Returns exactly 10 items (or less if fewer match)
  • page_size=50 → Fetches 50 items per request, returns ALL items
  • limit=100, page_size=25 → Returns 100 items, fetching 25 per request (4 requests)
"""GSI pagination - limit vs page_size behavior."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import GlobalSecondaryIndex

client = get_default_client()


class User(Model):
    """User model with status GSI."""

    model_config = ModelConfig(table="users_gsi_pagination")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    email = StringAttribute()
    status = StringAttribute()
    created_at = StringAttribute()
    age = NumberAttribute()

    status_index = GlobalSecondaryIndex(
        index_name="status-index",
        partition_key="status",
        sort_key="created_at",
    )


async def main():
    # Create table with GSI
    if not await client.table_exists("users_gsi_pagination"):
        await client.create_table(
            "users_gsi_pagination",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            global_secondary_indexes=[
                {
                    "index_name": "status-index",
                    "hash_key": ("status", "S"),
                    "range_key": ("created_at", "S"),
                    "projection": "ALL",
                }
            ],
        )

    # Create 25 active users
    for i in range(25):
        await User(
            pk=f"USER#{i:03d}",
            sk="PROFILE",
            email=f"user{i}@example.com",
            status="active",
            created_at=f"2024-01-{i + 1:02d}",
            age=20 + i,
        ).save()

    # limit = total items to return (stops after N items)
    # page_size = items per DynamoDB request (controls pagination)

    # Example 1: Get exactly 10 active users
    users = [u async for u in User.status_index.query(partition_key="active", limit=10)]
    print(f"limit=10: Got {len(users)} users")

    # Example 2: Get all active users, fetching 5 per page
    count = 0
    async for _ in User.status_index.query(partition_key="active", page_size=5):
        count += 1
    print(f"page_size=5 (no limit): Got {count} users")

    # Example 3: Get 15 active users, fetching 5 per page (3 requests)
    users = [
        u
        async for u in User.status_index.query(
            partition_key="active",
            limit=15,
            page_size=5,
        )
    ]
    print(f"limit=15, page_size=5: Got {len(users)} users")

    # Example 4: Manual pagination for "load more" UI
    result = User.status_index.query(partition_key="active", limit=10, page_size=10)
    first_page = [u async for u in result]
    print(f"First page: {len(first_page)} items")

    if result.last_evaluated_key:
        result = User.status_index.query(
            partition_key="active",
            limit=10,
            page_size=10,
            last_evaluated_key=result.last_evaluated_key,
        )
        second_page = [u async for u in result]
        print(f"Second page: {len(second_page)} items")


asyncio.run(main())

Common mistake

If you only set limit, it also controls the DynamoDB page size. Use both limit and page_size when you want to control them separately.

Manual pagination

For "load more" buttons:

"""Manual pagination with GSI - for 'load more' buttons."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import StringAttribute
from pydynox.indexes import GlobalSecondaryIndex

client = get_default_client()


class User(Model):
    model_config = ModelConfig(table="users_manual_pagination")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    email = StringAttribute()
    status = StringAttribute()
    name = StringAttribute()

    status_index = GlobalSecondaryIndex("status-index", partition_key="status", sort_key="pk")


async def main():
    # Create table with GSI
    if not await client.table_exists("users_manual_pagination"):
        await client.create_table(
            "users_manual_pagination",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            global_secondary_indexes=[
                {
                    "index_name": "status-index",
                    "hash_key": ("status", "S"),
                    "range_key": ("pk", "S"),
                    "projection": "ALL",
                }
            ],
        )

    # Create 25 active users
    for i in range(25):
        await User(
            pk=f"USER#{i:03d}",
            sk="PROFILE",
            email=f"user{i}@example.com",
            status="active",
            name=f"User {i}",
        ).save()

    # First page
    result = User.status_index.query(partition_key="active", limit=10, page_size=10)

    print("First page:")
    async for user in result:
        print(f"  {user.email}")

    # Check if there are more results
    if result.last_evaluated_key:
        print("\n--- Loading more ---\n")

        # Next page using last_evaluated_key
        next_result = User.status_index.query(
            partition_key="active",
            limit=10,
            page_size=10,
            last_evaluated_key=result.last_evaluated_key,
        )

        print("Second page:")
        async for user in next_result:
            print(f"  {user.email}")


asyncio.run(main())

Async queries

GSI queries are async by default. Use async for to iterate:

async for user in User.email_index.query(partition_key="john@example.com"):
    print(user.name)

# With filter
async for user in User.status_index.query(
    partition_key="active",
    filter_condition=User.age >= 18,
):
    print(user.email)

# Get first result
user = await User.email_index.query(partition_key="john@example.com").first()

For sync code, use sync_query:

for user in User.email_index.sync_query(partition_key="john@example.com"):
    print(user.name)

Metrics

Access query metrics using class methods:

async for user in User.email_index.query(partition_key="john@example.com"):
    print(user.name)

# Get last operation metrics
last = User.get_last_metrics()
if last:
    print(f"Duration: {last.duration_ms}ms")
    print(f"RCU consumed: {last.consumed_rcu}")

For more details, see Observability.

Multi-attribute composite keys

DynamoDB supports up to 4 attributes per partition key and 4 per sort key in GSIs. This is useful for multi-tenant apps or complex access patterns.

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import GlobalSecondaryIndex


class Product(Model):
    model_config = ModelConfig(table="products")

    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    tenant_id = StringAttribute()
    region = StringAttribute()
    created_at = StringAttribute()
    item_id = StringAttribute()
    category = StringAttribute()
    subcategory = StringAttribute()
    name = StringAttribute()
    price = NumberAttribute()

    # Multi-attribute GSI: 2 partition keys + 2 sort keys
    location_index = GlobalSecondaryIndex(
        index_name="location-index",
        partition_key=["tenant_id", "region"],
        sort_key=["created_at", "item_id"],
    )

    # Multi-attribute GSI: 2 partition keys only
    category_index = GlobalSecondaryIndex(
        index_name="category-index",
        partition_key=["category", "subcategory"],
    )

Query multi-attribute GSI

All partition key attributes are required. Sort key attributes are optional.

# ruff: noqa: F821
# All partition key attributes are required
async for product in Product.location_index.query(
    tenant_id="ACME",
    region="us-east-1",
):
    print(f"{product.name}: ${product.price}")

# Query category index
async for phone in Product.category_index.query(
    category="electronics",
    subcategory="phones",
):
    print(phone.name)

# With filter
async for product in Product.location_index.query(
    tenant_id="ACME",
    region="us-east-1",
    filter_condition=Product.price >= 1000,
):
    print(f"Expensive: {product.name}")

When to use multi-attribute keys

Use case Example
Multi-tenant apps partition_key=["tenant_id", "entity_type"]
Hierarchical data partition_key=["country", "state"]
Time-series sort_key=["year", "month", "day"]
Composite sorting sort_key=["priority", "created_at"]

Tip

Multi-attribute keys avoid the need to create synthetic composite keys like tenant_id#region. DynamoDB handles the composition for you.

Create table with GSI

When creating tables programmatically, include GSI definitions:

"""Create table with GSI using DynamoDBClient."""

import asyncio

from pydynox import DynamoDBClient

client = DynamoDBClient()


async def main():
    # Create table with GSI (skip if already exists)
    if not await client.table_exists("users_with_gsi"):
        await client.create_table(
            "users_with_gsi",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            global_secondary_indexes=[
                {
                    "index_name": "email-index",
                    "hash_key": ("email", "S"),
                    "projection": "ALL",
                },
                {
                    "index_name": "status-index",
                    "hash_key": ("status", "S"),
                    "range_key": ("pk", "S"),
                    "projection": "ALL",
                },
            ],
        )


asyncio.run(main())

Multi-attribute GSI in create_table

Use partition_keys and sort_keys (plural) for multi-attribute keys:

from pydynox import DynamoDBClient

client = DynamoDBClient()

# Create table with multi-attribute GSI (skip if already exists)
if not client.table_exists("products_multi_attr"):
    client.create_table(
        "products_multi_attr",
        partition_key=("pk", "S"),
        sort_key=("sk", "S"),
        global_secondary_indexes=[
            {
                "index_name": "location-index",
                "partition_keys": [("tenant_id", "S"), ("region", "S")],
                "sort_keys": [("created_at", "S"), ("item_id", "S")],
                "projection": "ALL",
            },
            {
                "index_name": "category-index",
                "partition_keys": [("category", "S"), ("subcategory", "S")],
                "projection": "ALL",
            },
        ],
    )

Projection types

Control which attributes are copied to the index:

Projection Description Use when
"ALL" All attributes (default) You need all data from the index
"KEYS_ONLY" Only key attributes You just need to check existence
"INCLUDE" Specific attributes You need some attributes, not all 
# Keys only - smallest index, lowest cost
{
    "index_name": "status-index",
    "partition_key": ("status", "S"),
    "projection": "KEYS_ONLY",
}

# Include specific attributes
{
    "index_name": "email-index",
    "partition_key": ("email", "S"),
    "projection": "INCLUDE",
    "non_key_attributes": ["name", "created_at"],
}

GSI limitations

  • GSIs are read-only. To update data, update the main table.
  • GSI queries are eventually consistent by default.
  • Each table can have up to 20 GSIs.
  • Multi-attribute keys: max 4 attributes per partition key, 4 per sort key.

Local secondary indexes

LSIs let you query by the same hash key but with a different sort key. They must be created when the table is created and cannot be added later.

Key features

  • Same hash key as the table, different sort key
  • Supports strongly consistent reads (unlike GSIs)
  • Must be defined at table creation time
  • Maximum 5 LSIs per table

When to use LSI vs GSI

Feature LSI GSI
Hash key Same as table Any attribute
Sort key Different from table Any attribute
Consistent reads Yes No
Add after table creation No Yes
Max per table 5 20

Use LSI when:

  • You need to query by the same hash key with different sort orders
  • You need strongly consistent reads on the index
  • You know the access patterns at table creation time

Define an LSI

"""Basic LSI definition - query by same hash key with different sort key."""

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import LocalSecondaryIndex


class Order(Model):
    """Order model with LSI for querying by status."""

    model_config = ModelConfig(table="orders")

    # Primary key: customer_id (hash) + order_id (range)
    customer_id = StringAttribute(partition_key=True)
    order_id = StringAttribute(sort_key=True)

    # Other attributes
    status = StringAttribute()
    total = NumberAttribute()
    created_at = StringAttribute()

    # LSI: query orders by customer_id + status
    # Same hash key (customer_id), different sort key (status)
    status_index = LocalSecondaryIndex(
        index_name="status-index",
        sort_key="status",
    )

    # LSI: query orders by customer_id + created_at
    created_index = LocalSecondaryIndex(
        index_name="created-index",
        sort_key="created_at",
    )

Query an LSI

LSI queries use partition_key= for the hash key (same as the table's hash key):

"""Query LSI - find orders by customer with different sort keys."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import LocalSecondaryIndex

client = get_default_client()


class Order(Model):
    """Order model with LSI."""

    model_config = ModelConfig(table="orders_lsi")

    customer_id = StringAttribute(partition_key=True)
    order_id = StringAttribute(sort_key=True)
    status = StringAttribute()
    total = NumberAttribute()
    created_at = StringAttribute()

    # LSI for querying by status
    status_index = LocalSecondaryIndex(
        index_name="status-index",
        sort_key="status",
    )


async def main():
    # Create table with LSI
    if not await client.table_exists("orders_lsi"):
        await client.create_table(
            "orders_lsi",
            partition_key=("customer_id", "S"),
            sort_key=("order_id", "S"),
            local_secondary_indexes=[
                {
                    "index_name": "status-index",
                    "range_key": ("status", "S"),
                    "projection": "ALL",
                }
            ],
        )

    # Create some orders
    await Order(
        customer_id="CUST#1",
        order_id="ORD#001",
        status="pending",
        total=100,
        created_at="2024-01-01",
    ).save()
    await Order(
        customer_id="CUST#1",
        order_id="ORD#002",
        status="shipped",
        total=250,
        created_at="2024-01-02",
    ).save()
    await Order(
        customer_id="CUST#1",
        order_id="ORD#003",
        status="pending",
        total=75,
        created_at="2024-01-03",
    ).save()

    # Query all orders for customer (using main table)
    print("All orders for CUST#1:")
    async for order in Order.query(partition_key="CUST#1"):
        print(f"  {order.order_id}: {order.status} - ${order.total}")

    # Query orders by status using LSI (partition_key= is the recommended way)
    print("\nPending orders for CUST#1 (via LSI):")
    async for order in Order.status_index.query(
        partition_key="CUST#1", sort_key_condition=Order.status == "pending"
    ):
        print(f"  {order.order_id}: ${order.total}")


asyncio.run(main())

Consistent reads

LSIs support strongly consistent reads. This is a key difference from GSIs.

"""LSI with consistent read - LSIs support strongly consistent reads."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import LocalSecondaryIndex

client = get_default_client()


class Order(Model):
    """Order model with LSI."""

    model_config = ModelConfig(table="orders_consistent")

    customer_id = StringAttribute(partition_key=True)
    order_id = StringAttribute(sort_key=True)
    status = StringAttribute()
    total = NumberAttribute()

    status_index = LocalSecondaryIndex(
        index_name="status-index",
        sort_key="status",
    )


async def main():
    # Create table
    if not await client.table_exists("orders_consistent"):
        await client.create_table(
            "orders_consistent",
            partition_key=("customer_id", "S"),
            sort_key=("order_id", "S"),
            local_secondary_indexes=[
                {
                    "index_name": "status-index",
                    "range_key": ("status", "S"),
                    "projection": "ALL",
                }
            ],
        )

    # Create an order
    await Order(
        customer_id="CUST#1",
        order_id="ORD#001",
        status="pending",
        total=100,
    ).save()

    # Query with consistent read (LSI-specific feature)
    # GSIs do NOT support consistent reads, but LSIs do!
    results = [
        o
        async for o in Order.status_index.query(
            partition_key="CUST#1",
            consistent_read=True,  # Strongly consistent read
        )
    ]

    print(f"Found {len(results)} orders with consistent read")


asyncio.run(main())

Pagination

LSI queries support the same pagination parameters as table queries and GSI queries:

Parameter What it does DynamoDB behavior
limit Max total items to return Stops iteration after N items
page_size Items per DynamoDB request Passed as Limit to DynamoDB API 
"""LSI pagination - limit vs page_size behavior."""

import asyncio

from pydynox import Model, ModelConfig, get_default_client
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import LocalSecondaryIndex

client = get_default_client()


class Order(Model):
    """Order model with LSI on created_at."""

    model_config = ModelConfig(table="orders_lsi_pagination")

    pk = StringAttribute(partition_key=True)  # customer_id
    sk = StringAttribute(sort_key=True)  # order_id
    created_at = StringAttribute()
    total = NumberAttribute()
    status = StringAttribute()

    created_at_index = LocalSecondaryIndex(
        index_name="created_at-index",
        sort_key="created_at",
    )


async def main():
    # Create table with LSI
    if not await client.table_exists("orders_lsi_pagination"):
        await client.create_table(
            "orders_lsi_pagination",
            partition_key=("pk", "S"),
            sort_key=("sk", "S"),
            local_secondary_indexes=[
                {
                    "index_name": "created_at-index",
                    "range_key": ("created_at", "S"),
                    "projection": "ALL",
                }
            ],
        )

    # Create 25 orders for one customer
    for i in range(25):
        await Order(
            pk="CUSTOMER#1",
            sk=f"ORDER#{i:03d}",
            created_at=f"2024-01-{i + 1:02d}",
            total=100 + i * 10,
            status="pending" if i % 2 == 0 else "shipped",
        ).save()

    # limit = total items to return (stops after N items)
    # page_size = items per DynamoDB request (controls pagination)

    # Example 1: Get exactly 10 orders for a customer
    orders = [
        o
        async for o in Order.created_at_index.query(
            partition_key="CUSTOMER#1",
            limit=10,
        )
    ]
    print(f"limit=10: Got {len(orders)} orders")

    # Example 2: Get all orders, fetching 5 per page
    count = 0
    async for _ in Order.created_at_index.query(
        partition_key="CUSTOMER#1",
        page_size=5,
    ):
        count += 1
    print(f"page_size=5 (no limit): Got {count} orders")

    # Example 3: Get 15 orders, fetching 5 per page (3 requests)
    orders = [
        o
        async for o in Order.created_at_index.query(
            partition_key="CUSTOMER#1",
            limit=15,
            page_size=5,
        )
    ]
    print(f"limit=15, page_size=5: Got {len(orders)} orders")

    # Example 4: Manual pagination with consistent reads (LSI supports this!)
    result = Order.created_at_index.query(
        partition_key="CUSTOMER#1",
        limit=10,
        page_size=10,
        consistent_read=True,
    )
    first_page = [o async for o in result]
    print(f"First page (consistent): {len(first_page)} items")

    if result.last_evaluated_key:
        result = Order.created_at_index.query(
            partition_key="CUSTOMER#1",
            limit=10,
            page_size=10,
            consistent_read=True,
            last_evaluated_key=result.last_evaluated_key,
        )
        second_page = [o async for o in result]
        print(f"Second page (consistent): {len(second_page)} items")


asyncio.run(main())

LSI projections

Like GSIs, LSIs support different projection types:

"""LSI with different projection types."""

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import LocalSecondaryIndex


class Order(Model):
    """Order model with LSIs using different projections."""

    model_config = ModelConfig(table="orders_projections")

    customer_id = StringAttribute(partition_key=True)
    order_id = StringAttribute(sort_key=True)
    status = StringAttribute()
    total = NumberAttribute()
    notes = StringAttribute()

    # ALL projection - includes all attributes
    status_index = LocalSecondaryIndex(
        index_name="status-index",
        sort_key="status",
        projection="ALL",
    )

    # KEYS_ONLY projection - only key attributes
    # Smaller index, faster queries when you only need keys
    total_index = LocalSecondaryIndex(
        index_name="total-index",
        sort_key="total",
        projection="KEYS_ONLY",
    )

    # INCLUDE projection - keys + specific attributes
    # Balance between size and data availability
    notes_index = LocalSecondaryIndex(
        index_name="notes-index",
        sort_key="notes",
        projection=["status", "total"],  # Include these non-key attributes
    )

Create table with LSI

Using DynamoDBClient:

"""Create table with LSI using DynamoDBClient."""

import asyncio

from pydynox import DynamoDBClient

client = DynamoDBClient()


async def main():
    # Create table with LSI (skip if already exists)
    if not await client.table_exists("orders_with_lsi"):
        await client.create_table(
            "orders_with_lsi",
            partition_key=("customer_id", "S"),
            sort_key=("order_id", "S"),
            local_secondary_indexes=[
                {
                    "index_name": "status-index",
                    "range_key": ("status", "S"),
                    "projection": "ALL",
                },
                {
                    "index_name": "created-index",
                    "range_key": ("created_at", "S"),
                    "projection": "KEYS_ONLY",
                },
            ],
        )


asyncio.run(main())

Using Model.create_table():

"""Create table with LSI using Model.create_table()."""

import asyncio

from pydynox import DynamoDBClient, Model, ModelConfig, set_default_client
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.indexes import LocalSecondaryIndex

# Setup client
client = DynamoDBClient()
set_default_client(client)


class Order(Model):
    """Order model with LSI - table created automatically from model definition."""

    model_config = ModelConfig(table="orders_model_lsi")

    customer_id = StringAttribute(partition_key=True)
    order_id = StringAttribute(sort_key=True)
    status = StringAttribute()
    total = NumberAttribute()

    # LSI defined on model
    status_index = LocalSecondaryIndex(
        index_name="status-index",
        sort_key="status",
    )


async def main():
    # Create table from model definition
    # LSIs are automatically included!
    if not await Order.table_exists():
        await Order.create_table(wait=True)
        print("Table created with LSI")

    # Use the model
    await Order(
        customer_id="CUST#1",
        order_id="ORD#001",
        status="pending",
        total=100,
    ).save()

    # Query using LSI
    async for order in Order.status_index.query(partition_key="CUST#1"):
        print(f"Order: {order.order_id} - {order.status}")


asyncio.run(main())


asyncio.run(main())

LSI limitations

  • Must be created with the table (cannot add later)
  • Maximum 5 LSIs per table
  • Hash key must be the same as the table's hash key
  • Table must have a range key to use LSIs

Next steps

  • Conditions - Filter and conditional writes
  • Query - Query items by hash key with conditions
  • Tables - Create tables with indexes