trustcafe-api-wrapper/docs/CUSTOM_REQUESTS.md

Custom Requests Guide

This guide covers making advanced API calls using the TrustCafé API wrapper's direct request functionality. This is for when jobs or wrappers don't provide the level of control you need.

Table of Contents

• Overview
• Basic Usage
• Authentication
• Common Patterns
• Advanced Usage
• Pagination
• Error Handling
• Examples

Overview

What is make_request?

make_request is the core method of the APIClient class that allows you to make raw API calls to TrustCafé endpoints directly. It bypasses jobs and wrappers, giving you complete control over:

• HTTP methods (GET, POST, PUT, DELETE)
• API endpoints
• Request paths
• Request bodies
• Query parameters
• Authentication

When to Use Custom Requests

Use make_request when:

• No wrapper or job exists for your operation
• You need a custom HTTP endpoint
• Building complex queries programmatically
• Using specialized API features
• Need maximum flexibility

Don't use when:

• Standard operation is covered by wrappers (too much boilerplate)
• Simpler option exists (jobs/wrappers)
• You're a beginner (wrappers/jobs are easier)

Basic Usage

Fundamentals

import trustcafeapiwrapper
import os

API = trustcafeapiwrapper.APIClient(
    client_id=os.getenv("CLIENT_ID"),
    client_secret=os.getenv("CLIENT_SECRET"),
    env="alpha",
    debug=True
)

API.handle_token()

The method signature

response = API.make_request(
    method="GET",
    endpoint="content",
    path="post/some-id",
    data=None,
    authenticate=True,
    query_params=None
)

Parameters Explained

Parameter	Type	Required	Default	Description
method	str	✅ Yes	-	HTTP method: "GET", "POST", "PUT", "DELETE"
endpoint	str	✅ Yes	-	API service: "content", "auth", "audrey", etc.
path	str	✅ Yes	-	API path after endpoint
data	dict	❌ No	None	Request body for POST/PUT/DELETE
authenticate	bool	❌ No	True	Include authentication token
query_params	dict	❌ No	None	Query parameters for GET requests

Simple GET Request

# Get a post
response = API.make_request(
    "GET",
    "content",
    "post/my-post-id"
)

print(response)

POST Request

# Create a new resource
response = API.make_request(
    "POST",
    "content",
    "post/new",
    data={
        "postText": "My custom post",
        "parent": {
            "pk": "maintrunk#maintrunk",
            "sk": "maintrunk#maintrunk"
        }
    }
)

print(response)

PUT Request

# Update an existing resource
response = API.make_request(
    "PUT",
    "content",
    "post/my-post-id",
    data={
        "pk": "my-post-id",
        "sk": "my-post-id",
        "newPostText": "Updated content"
    }
)

print(response)

DELETE Request

# Delete a resource
response = API.make_request(
    "DELETE",
    "content",
    "post/my-post-id"
)

print(response)

Authentication

By default, authentication is automatic:

# This includes the access token in the Authorization header
response = API.make_request(
    "GET",
    "content",
    "post/my-post-id",
    authenticate=True  # Default
)

Making guest requests (no authentication):

# Useful for public endpoints
response = API.make_request(
    "GET",
    "content",
    "post/public-post",
    authenticate=False  # No token needed
)

Manual token injection (advanced):

# Sometimes you may need to bypass the token management
response = API.make_request(
    "GET",
    "content",
    "post/some-path",
    headers={
        "Authorization": "Bearer your-token-manually-set"
    }
)

Common Patterns

Pagination

TrustCafé responses may include pagination for large result sets:

def get_all_posts(limit=100):
    """Fetch all posts with pagination"""

    def fetch_page(start_key=None):
        """Fetch a single page of posts"""
        params = {"Limit": limit}

        if start_key:
            params["ExclusiveStartKey"] = start_key

        response = API.make_request(
            "GET",
            "content",
            "post/listall",
            query_params=params,
            authenticate=True
        )

        return response

    all_posts = []
    start_key = None

    while True:
        page = fetch_page(start_key)

        items = page.get('Items', [])
        all_posts.extend(items)

        # Check if there are more pages
        has_more = 'LastEvaluatedKey' in page
        if not has_more or len(items) == 0:
            break

        # Set start key for next page
        start_key = page['LastEvaluatedKey']

    return all_posts


# Usage
API.handle_token()
all_posts = get_all_posts(limit=50)
print(f"Total posts: {len(all_posts)}")

Filtering with Query Parameters

# Get posts created after a specific date
response = API.make_request(
    "GET",
    "content",
    "post/listpublic",
    query_params={
        "creationDateAfter": "2024-01-01T00:00:00Z"
    },
    authenticate=True
)

# Get posts in a specific branch
response = API.make_request(
    "GET",
    "content",
    "post/ref-subwiki/music",
    query_params={
        "maxResults": 20,
        "showDeleted": False
    },
    authenticate=True
)

Building Dynamic Paths

def get_post_by_slug(slug):
    """Get a post by slugified ID"""
    response = API.make_request(
        "GET",
        "content",
        f"post/id/{slug}",
        authenticate=True
    )
    return response

# Usage
post = get_post_by_slug("my-custom-slug-123")

def get_comments_for_post(post_id):
    """Get comments for a specific post"""
    response = API.make_request(
        "GET",
        "content",
        f"comment/bypostid/{post_id}",
        authenticate=True
    )
    return response

Advanced Usage

Multiple Endpoints

TrustCafé has multiple API service endpoints:

# Content API
content_response = API.make_request(
    "GET",
    "content",
    "post/listall",
    query_params={"Limit": 10}
)

# Auth API (for tokens)
auth_response = API.make_request(
    "POST",
    "auth",
    "token",
    data={
        "client_id": "your-id",
        "client_secret": "your-secret",
        "grant_type": "client_credentials"
    },
    authenticate=False
)

# Audrey API (various services)
audrey_response = API.make_request(
    "GET",
    "audrey",
    "some/endpoint",
    authenticate=True
)

Complex Payloads

# Building complex payloads for updates
def update_post_completion(post_id, completion_status):
    """Update post with new completion status"""
    response = API.make_request(
        "PUT",
        "content",
        f"post/{post_id}",
        data={
            "pk": post_id,
            "sk": post_id,
            "postText": "Updated post with completion status",
            "completion": completion_status
        },
        authenticate=True
    )
    return response


# Usage
update_post_completion(
    "current-post-pk",
    {
        "isComplete": True,
        "completedBy": "username",
        "completedAt": "2024-01-01"
    }
)

Batch Operations

def batch_create_posts(posts_data):
    """Create multiple posts in a single batch"""

    # Build batch request
    batch_data = {
        "Post": [
            {
                "postText": post["text"],
                "parent": {
                    "pk": post["parent_pk"],
                    "sk": post["parent_sk"]
                }
            }
            for post in posts_data
        ]
    }

    response = API.make_request(
        "POST",
        "content",
        "post/batch-create",
        data=batch_data,
        authenticate=True
    )

    return response

Error Handling for Custom Requests

def safe_api_request(method, endpoint, path, data=None):
    """
    Safely execute an API request with proper error handling.
    """
    try:
        response = API.make_request(
            method,
            endpoint,
            path,
            data=data,
            authenticate=True
        )

        # Check for unexpected errors
        if 'error' in response:
            raise Exception(f"API Error: {response['error']}")

        return response

    except Exception as e:
        print(f"Request failed: {e}")
        # Handle error: retry, notify, log, etc.
        raise

Pagination

Pagination Strategy

When handling paginated results, always check for pagination metadata:

def fetch_all_content():
    """Fetch all content with pagination handling"""

    all_content = []
    next_token = None

    while True:
        params = {
            "Limit": 100
        }

        if next_token:
            params["ExclusiveStartKey"] = next_token

        try:
            response = API.make_request(
                "GET",
                "content",
                "content/list",
                query_params=params,
                authenticate=True
            )

        except Exception as e:
            print(f"Error fetching page: {e}")
            break

        # Process current page
        items = response.get('Items', [])
        all_content.extend(items)

        # Check for next page
        has_more = 'LastEvaluatedKey' in response

        if not has_more or len(items) < 100:
            print(f"Total items retrieved: {len(all_content)}")
            break

        next_token = response['LastEvaluatedKey']
        print(f"Fetching next page (token: {next_token[:10]}...)")

    return all_content

Processing Large Datasets

def process_content_in_batches(batch_size=100):
    """Process content in manageable batches"""

    batch_number = 0

    while True:
        batch_number += 1
        print(f"Processing batch {batch_number}...")

        try:
            response = API.make_request(
                "GET",
                "content",
                "content/list",
                query_params={
                    "Limit": batch_size
                },
                authenticate=True
            )

            items = response.get('Items', [])

            if not items:
                print("No more items to process")
                break

            # Process each item in the batch
            for item in items:
                # Your processing logic here
                process_item(item)

            # Check for more items
            has_more = 'LastEvaluatedKey' in response
            if not has_more:
                break

        except Exception as e:
            print(f"Error in batch {batch_number}: {e}")
            # Retry logic could go here
            break

    print(f"Completed processing {batch_number} batches")

Error Handling

Common Error Scenarios

def handle_api_request(method, endpoint, path, data=None):
    """Robust API request handler"""

    try:
        # Make request
        response = API.make_request(
            method,
            endpoint,
            path,
            data=data,
            authenticate=True
        )

        # Check for API-level errors
        if 'error' in response:
            raise APIError(
                endpoint=endpoint,
                path=path,
                api_error=response['error'],
                message=response.get('message', '')
            )

        return response

    except APIError as e:
        print(f"API Error: {e}")
        # Retry logic or fallback could go here
        raise

    except Exception as e:
        print(f"Unexpected error: {e}")
        # Log the error
        log_error(e)
        # Could retry or use fallback
        raise

Retry Logic

import time
import logging

logger = logging.getLogger(__name__)

def retry_on_failure(request_func, max_retries=3, retry_delay=1):
    """
    Retry a request function on failure.

    Args:
        request_func: Function that makes the API request
        max_retries: Maximum number of retry attempts
        retry_delay: Seconds to wait between retries

    Returns:
        The successful response
    """
    for attempt in range(max_retries):
        try:
            return request_func()

        except Exception as e:
            logger.warning(
                f"Request failed (attempt {attempt + 1}/{max_retries}): {e}"
            )

            if attempt < max_retries - 1:
                logger.info(f"Retrying in {retry_delay} seconds...")
                time.sleep(retry_delay)
            else:
                logger.error("Max retries reached. Giving up.")
                raise

    # This line should never be reached due to raise above
    raise Exception("Unexpected state in retry function")

Examples

Example 1: Complex Query

def search_posts_advanced(search_term, branch="all", limit=50):
    """
    Advanced search for posts with filtering.

    Note: This is an EXAMPLE - you'll need to adjust to match
    TrustCafé's actual search/query API structure.
    """

    # Build search query
    query_data = {
        "search": {
            "term": search_term,
            "filters": {
                "branches": [branch] if branch != "all" else None
            }
        }
    }

    response = API.make_request(
        "POST",
        "content",
        "search",
        data=query_data,
        authenticate=True
    )

    return response

Example 2: Custom Analytics Query

def get_post_engagement_metrics(post_id):
    """
    Get engagement metrics for a post.

    Note: This requires mapping to TrustCafé's actual endpoint
    and data structure for engagement metrics.
    """

    response = API.make_request(
        "GET",
        "content",
        f"post/{post_id}/analytics",
        query_params={
            "metrics": [
                "views",
                "likes",
                "comments",
                "shares"
            ],
            "timeRange": "7d"
        },
        authenticate=True
    )

    return response

Example 3: Multi-step Transaction

def create_post_and_related_content(post_text, related_items):
    """
    Create a post and related content in a transaction-like operation.
    """
    # Step 1: Create the main post
    post_response = API.make_request(
        "POST",
        "content",
        "post",
        data={
            "postText": post_text,
            "parent": {
                "pk": "maintrunk#maintrunk",
                "sk": "maintrunk#maintrunk"
            }
        },
        authenticate=True
    )

    post_id = post_response.get('pk')

    # Step 2: Create related content (if applicable)
    if related_items:
        for item in related_items:
            API.make_request(
                "POST",
                "content",
                "related-item",
                data={
                    "postId": post_id,
                    "externalId": item['id'],
                    "type": item['type'],
                    "url": item['url']
                },
                authenticate=True
            )

    return post_response

Example 4: Conditional Operations

def conditional_update_post(post_id, updates, condition=None):
    """
    Update a post only if a condition is met.

    Note: This maps to CousinDB/GSI-style conditional updates.
    Implementation depends on TrustCafé's specific API.
    """

    update_data = {
        "pk": post_id,
        "sk": post_id,
        **updates
    }

    if condition:
        update_data["condition"] = {
            "expression": condition["expression"],
            "values": condition["values"]
        }

    response = API.make_request(
        "PUT",
        "content",
        f"post/{post_id}",
        data=update_data,
        authenticate=True
    )

    return response

---

Summary

make_request provides maximum flexibility for making API calls, but with greater complexity comes the need for careful:

• Error handling - Always wrap in try/except
• Logging - Track request/response for debugging
• Testing - Verify your custom endpoints
• Documentation - Document any custom patterns you discover

For most use cases, wrappers and jobs provide the right balance of simplicity and control. Use make_request when standard wrappers don't fit your needs.

---

Next Steps

• API Reference - Complete list of jobs
• Wrappers Guide - High-level wrappers for common operations
• Examples - Real-world usage examples
• Troubleshooting - Common issues and solutions