Skip to main content
/ Docs
Quickstart API SDKs

Python SDK: Advanced

Async operations, error handling, interceptors, and best practices for the Python SDK.

Updated

Overview

This section covers advanced topics including async operations, error handling, interceptors, pagination, retry configuration, and type hints.

Async Support

The SDK provides a full async client for non-blocking operations:

async_example.py 
import asyncio
from trix import AsyncTrix

async def main():
    async with AsyncTrix(api_key="gdb_your_api_key") as client:
        memory = await client.memories.create(
            content="Async memory creation"
        )

        memories = await asyncio.gather(
            client.memories.create(content="First"),
            client.memories.create(content="Second"),
            client.memories.create(content="Third")
        )

        async for memory in client.memories.iter(page_size=100):
            print(memory.content)

        context = await client.graph.get_context(
            query="Python async patterns",
            depth=2
        )

asyncio.run(main())

Async from Environment

async_from_env.py 
from trix import AsyncTrix

client = AsyncTrix.from_env()

async def process():
    async with client:
        memories = await client.memories.list(limit=10)
        return memories

Common Async Patterns

async_patterns.py 
import asyncio
from trix import AsyncTrix

async def process_memories(client, memory_ids):
    tasks = [client.memories.get(mid) for mid in memory_ids]
    return await asyncio.gather(*tasks)

async def search_with_timeout(client, query, timeout=5.0):
    try:
        return await asyncio.wait_for(
            client.memories.list(q=query),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        print("Search timed out")
        return None

async def batch_create(client, contents, batch_size=10):
    for i in range(0, len(contents), batch_size):
        batch = contents[i:i + batch_size]
        tasks = [client.memories.create(content=c) for c in batch]
        await asyncio.gather(*tasks)
        print(f"Created batch {i // batch_size + 1}")

Error Handling

The SDK provides typed exception classes for precise error handling:

error_handling.py 
from trix import (
    TrixError,
    AuthenticationError,
    PermissionError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ConnectionError,
    TimeoutError,
    ServerError
)

try:
    memory = client.memories.create(content="")
except ValidationError as e:
    print(f"Validation failed: {e}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after} seconds")
except AuthenticationError as e:
    print("Invalid API key")
except NotFoundError as e:
    print("Resource not found")
except PermissionError as e:
    print("Insufficient permissions")
except ConnectionError as e:
    print(f"Network error: {e}")
except TimeoutError as e:
    print("Request timed out")
except ServerError as e:
    print(f"Server error: {e}")
except TrixError as e:
    print(f"Trix error: {e}")

Error Properties

Error TypeProperties
RateLimitErrorretry_after (seconds to wait)
ValidationErrorerrors (field-level errors)
ServerErrorstatus_code, response
All errorsmessage

Handling Rate Limits

rate_limit_handling.py 
import time
from trix import Trix, RateLimitError

def create_with_retry(client, content, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.memories.create(content=content)
        except RateLimitError as e:
            if attempt < max_retries - 1:
                print(f"Rate limited, waiting {e.retry_after}s...")
                time.sleep(e.retry_after)
            else:
                raise

Interceptors

Interceptors let you hook into the request/response lifecycle for logging, authentication, or custom logic.

Request Interceptor

request_interceptor.py 
from trix import RequestContext

def log_request(ctx: RequestContext) -> RequestContext:
    print(f"Request: {ctx.method} {ctx.url}")
    ctx.headers["X-Request-ID"] = generate_request_id()
    return ctx

remove = client.add_request_interceptor(log_request)
remove()

Response Interceptor

response_interceptor.py 
from trix import ResponseContext

def log_response(ctx: ResponseContext) -> ResponseContext:
    print(f"Response: {ctx.status_code}")
    if ctx.headers.get("x-response-time", 0) > 1000:
        print("Warning: Slow response detected")
    return ctx

client.add_response_interceptor(log_response)

Error Interceptor

error_interceptor.py 
from trix import RequestContext

def handle_error(error: Exception, ctx: RequestContext) -> Exception:
    print(f"Error on {ctx.method} {ctx.url}: {error}")
    error_tracker.capture_exception(error, extra={"request": ctx})
    return error

client.add_error_interceptor(handle_error)

Interceptors are useful for logging, metrics, authentication token refresh, and custom error handling across all API calls.

Pagination

List endpoints return paginated responses. Use iter() methods for the simplest approach, or handle pagination manually.

Automatic Iteration (Recommended)

auto_pagination.py 
for memory in client.memories.iter(page_size=100):
    process_memory(memory)

async for memory in client.memories.iter(page_size=100):
    await process_memory(memory)

for memory in client.memories.iter(page_size=50, max_items=500):
    process_memory(memory)

for cluster in client.clusters.iter():
    pass

for webhook in client.webhooks.list():
    pass

Manual Pagination

manual_pagination.py 
result = client.memories.list(limit=100, offset=0)

print(f"Total: {result.total}")
print(f"Returned: {len(result.data)}")

offset = 0
limit = 100

while True:
    result = client.memories.list(limit=limit, offset=offset)
    for memory in result.data:
        process_memory(memory)
    if len(result.data) < limit:
        break
    offset += limit

Retry & Backoff

The SDK automatically retries failed requests with exponential backoff.

Default Retry Behavior

default_retry.py 
{
    "max_retries": 3,
    "initial_delay": 1.0,
    "max_delay": 60.0,
    "exponential_base": 2.0,
    "jitter": True
}

# Retryable errors:
# - RateLimitError (429)
# - ConnectionError
# - TimeoutError
# - ServerError (5xx)

Custom Retry Configuration

custom_retry.py 
from trix import Trix, RetryConfig

retry_config = RetryConfig(
    max_retries=5,
    initial_delay=2.0,
    max_delay=120.0,
    exponential_base=2.0,
    jitter=True
)

client = Trix(
    api_key="gdb_your_api_key",
    retry_config=retry_config
)

Type Hints

The SDK provides comprehensive Pydantic models for full type safety:

types.py 
from trix import (
    Memory, MemoryType, MemoryCreate, MemoryUpdate, MemoryList,
    Relationship, RelationshipType, RelationshipCreate,
    Cluster, ClusterCreate,
    Space, SpaceCreate,
    SearchResult, SearchResults, SearchMode,
    AgentSession, SessionMemory, ConsolidationStrategy, ConsolidationResult,
    GraphNode, GraphTraversal, GraphContext, ShortestPath, Direction,
    Webhook, WebhookEvent, WebhookDelivery,
    Job, JobStatus, JobStats,
    Highlight, ExtractionType,
    FeedbackResult,
    RetryConfig, PoolConfig
)

Memory Model

memory_model.py 
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
from datetime import datetime

class Memory(BaseModel):
    id: str
    content: str
    type: MemoryType
    tags: Optional[List[str]] = None
    metadata: Optional[Dict[str, Any]] = None
    priority: Optional[int] = None
    space_id: Optional[str] = None
    embedding: Optional[List[float]] = None
    access_count: int = 0
    created_at: datetime
    updated_at: datetime
    transcript: Optional[str] = None
    audio_url: Optional[str] = None

Relationship Model

relationship_model.py 
class Relationship(BaseModel):
    id: str
    source_id: str
    target_id: str
    relationship_type: RelationshipType
    description: Optional[str] = None
    weight: float = 1.0
    bidirectional: bool = False
    reinforcement_count: int = 0
    created_at: datetime
    updated_at: datetime

Best Practices

Use Context Managers

Always use context managers to ensure proper resource cleanup:

context_managers.py 
with Trix.from_env() as client:
    memory = client.memories.create(content="Hello")

async with AsyncTrix.from_env() as client:
    memory = await client.memories.create(content="Hello")

Handle Errors Gracefully

error_handling_best.py 
from trix import TrixError, ValidationError, RateLimitError

def safe_create_memory(client, content, tags=None):
    try:
        return client.memories.create(content=content, tags=tags or [])
    except ValidationError as e:
        print(f"Invalid input: {e}")
        return None
    except RateLimitError as e:
        print(f"Rate limited, try again in {e.retry_after}s")
        return None
    except TrixError as e:
        print(f"API error: {e}")
        return None

Use Bulk Operations for Efficiency

bulk_operations.py 
from trix import MemoryCreate

memories = client.memories.bulk_create([
    MemoryCreate(content=c) for c in contents
])

Use Iterators for Large Datasets

iterators.py 
for memory in client.memories.iter(page_size=100):
    process_memory(memory)

Configure Timeouts Appropriately

timeouts.py 
from trix import Trix

client = Trix(api_key="gdb_your_api_key", timeout=10.0)

client = Trix(api_key="gdb_your_api_key", timeout=120.0)

Complete Example

Here's a complete example demonstrating common SDK operations:

complete_example.py 
from trix import (
    Trix, MemoryType, RelationshipType,
    ConsolidationStrategy, ValidationError, RateLimitError
)

def main():
    with Trix.from_env() as client:
        try:
            memory1 = client.memories.create(
                content="Python is excellent for data science",
                type=MemoryType.TEXT,
                tags=["python", "data-science"]
            )
            memory2 = client.memories.create(
                content="Pandas provides powerful data manipulation",
                type=MemoryType.TEXT,
                tags=["python", "pandas"]
            )
            print(f"Created memories: {memory1.id}, {memory2.id}")

            relationship = client.relationships.create(
                source_id=memory1.id,
                target_id=memory2.id,
                relationship_type=RelationshipType.RELATED_TO,
                weight=1.5
            )
            print(f"Created relationship: {relationship.id}")

            cluster = client.clusters.create(
                name="Python Data Science",
                color="#3776AB"
            )
            client.clusters.add_memory(cluster.id, memory1.id)
            client.clusters.add_memory(cluster.id, memory2.id)
            print(f"Created cluster: {cluster.id}")

            similar = client.search.similar(memory1.id, limit=5)
            for result in similar:
                print(f"  {result.score:.2f}: {result.memory.content[:40]}...")

            context = client.graph.get_context(query="Python data analysis", depth=2)
            print(f"Context includes {len(context.memories)} related memories")

            session = client.agent.create_session(
                session_id="learning_session",
                metadata={"topic": "data science"}
            )
            client.agent.add_session_memory(
                session_id="learning_session",
                content="User is learning Python for data science",
                role="system"
            )
            client.agent.end_session(
                session_id="learning_session",
                summary="Explored Python data science tools"
            )
            print("Session completed")

            client.feedback.quick(memory_id=memory1.id, useful=True)
            print("Feedback submitted")

        except ValidationError as e:
            print(f"Validation error: {e}")
        except RateLimitError as e:
            print(f"Rate limited. Retry after: {e.retry_after}")
        except Exception as e:
            raise

if __name__ == "__main__":
    main()

Next Steps