trustcafe-api-wrapper/docs/API_REFERENCE.md

TrustCafé API Reference

Complete reference guide for all available jobs in the TrustCafé API wrapper.

Table of Contents

• Preparation
• Authentication Jobs
• Post Jobs
• Comment Jobs
• UserProfile Jobs
• Follow Jobs
• Vote Jobs
• Reaction Jobs
• Notification Jobs
• Trust Jobs
• Branch Jobs
• Feed Jobs
• Block Jobs
• Mute Jobs
• Error Handling

Preparation

Before using jobs, initialize the API client:

import trustcafeapiwrapper
import os

API = trustcafeapiwrapper.APIClient(
    client_id=os.getenv("TRUSTCAFE_CLIENT_ID"),
    client_secret=os.getenv("TRUSTCAFE_CLIENT_SECRET"),
    environment="alpha",  # or "production"
    debug=False
)

# Authenticate
API.handle_token()

Authentication Jobs

sign_in()

Obtains a new access token using OAuth2 client-credentials flow.

token_data = API.sign_in()

Returns:

{
    "access_token": "string (3-month duration)",
    "access_token_timeout": "integer (Unix timestamp)"
}

Parameters: None

Error Cases:

• Invalid client credentials
• Incorrect environment
• Network errors

---

Post Jobs

Fetch, create, update, and manage posts.

post.get()

Fetches a specific post by ID or slug.

post_data = API.run_job('post.get', "post-slug-or-id")

Parameters:

• post_slug (str): The post identifier (slug or ID)

Returns: dict - Post data

Example:

post = API.run_job('post.get', "my-first-post")
print(post.get('postText'))

---

post.create()

Creates a new post.

post = API.run_job('post.create', payload)

Parameters:

• postText (str): The text content of the post
• parent (dict, required): Parent object containing pk and sk
  • pk (str): Partition key
  • sk (str): Sort key
• blurLabel (str, optional): Label for blurring content
• cardUrl (str, optional): URL for link preview card
• collaborative (bool, optional): Enable collaborative editing (default: False)

Returns: dict - Created post data

Example:

post = API.run_job('post.create', {
    "postText": "This is a new post",
    "parent": {
        "pk": "maintrunk#maintrunk",
        "sk": "maintrunk#maintrunk"
    }
})

---

post.update()

Updates an existing post.

updated_post = API.run_job('post.update', payload)

Parameters:

• pk (str): Post partition key
• sk (str): Post sort key
• newPostText (str, optional): New post content
• Other optional fields to update

Returns: dict - Updated post data

Example:

post = API.run_job('post.update', {
    "pk": "post-id-123",
    "sk": "post-id-123",
    "newPostText": "Updated post content"
})

Note: Full field list requires review of API documentation.

---

post.listall()

Lists all posts.

posts = API.run_job('post.listall')

Parameters:

• lastEvaluatedKey (dict, optional): Pagination key for next page

Returns: dict - List of all posts

Example:

all_posts = API.run_job('post.listall')
print(f"Total posts: {len(all_posts.get('Items', []))}")

---

post.listpublic()

Lists public posts only. This endpoint does not require authentication.

public_posts = API.run_job('post.listpublic')

Parameters:

• lastEvaluatedKey (dict, optional): Pagination key for next page

Returns: dict - List of public posts

Example:

public_posts = API.run_job('post.listpublic')
for post in public_posts.get('Items', []):
    print(post.get('pk'))

---

post.listbybranch()

Lists posts in a specific branch.

branch_posts = API.run_job('post.listbybranch', branch_slug)

Parameters:

• branch_slug (str): The branch/subwiki slug identifier
• lastEvaluatedKey (dict, optional): Pagination key for next page

Returns: dict - List of posts in the branch

Example:

music_posts = API.run_job('post.listbybranch', "music")

---

post.listbyuserprofile()

Lists posts by a specific user profile.

user_posts = API.run_job('post.listbyuserprofile', user_slug)

Parameters:

• user_slug (str): The user profile slug
• lastEvaluatedKey (dict, optional): Pagination key for next page

Returns: dict - List of posts by user

Example:

journals = API.run_job('post.listbyuserprofile', "philosopher-jon")

---

post.listremoved()

Lists removed/deleted posts.

removed = API.run_job('post.listremoved')

Parameters:

• lastEvaluatedKey (dict, optional): Pagination key for next page

Returns: dict - List of removed posts

Example:

removed_posts = API.run_job('post.listremoved')
for post in removed_posts.get('Items', []):
    print(post.get('pk'))

---

Comment Jobs

Manage comments on posts.

comment.create()

Creates a new comment.

comment = API.run_job('comment.create', payload)

Parameters:

• commentText (str): Comment content
• parent (dict, required): Parent object
  • pk (str): Partition key
  • sk (str): Sort key
• Other optional fields as API requires

Returns: dict - Created comment data

Example:

comment = API.run_job('comment.create', {
    "commentText": "Great post!",
    "parent": {
        "pk": "post-id",
        "sk": "post-id"
    }
})

---

comment.listtbypostid()

Lists comments for a specific post.

comments = API.run_job('comment.listtbypostid', post_id)

Parameters:

• post_id (str): The post ID

Returns: dict - List of comments

Example:

comments = API.run_job('comment.listtbypostid', "current-post-id")
for comment in comments.get('Items', []):
    print(comment.get('commentText'))

---

UserProfile Jobs

Retrieve user profile information.

userprofile.get()

Retrieves a user profile.

profile = API.run_job('userprofile.get', username)

Parameters:

• username (str): The username to retrieve

Returns: dict - User profile data

Example:

profile = API.run_job('userprofile.get', "philosopher-jon")
print(profile.get('username'))
print(profile.get('trustScore'))

---

Follow Jobs

Manage follow relationships.

follow.follow()

Follow or unfollow a user or subwiki.

result = API.run_job('follow.follow', payload)

Parameters:

• payload (dict): Follow payload containing:
  • isFollowing (bool): true to follow, false to unfollow
  • parent (dict): Object with pk and sk for the target
  • followType (str): Entity type ('userprofile' or 'subwiki')
  • parentSlug (str): Slug of the user or subwiki
  • preferences (dict, optional): Notification preferences

Returns: dict - Follow relationship data

Example:

# Follow a user
followed = API.run_job('follow.follow', {
    "isFollowing": True,
    "parent": {
        "pk": "userprofile#philosopher-jon",
        "sk": "userprofile#philosopher-jon"
    },
    "followType": "userprofile",
    "parentSlug": "philosopher-jon",
    "preferences": {
        "notification": True,
        "emailNew": False,
        "emailDigest": True
    }
})

---

Vote Jobs

Cast votes on posts/content.

vote.votecast()

Casts a vote on a post or comment.

vote = API.run_job('vote.votecast', payload)

Parameters:

• parent (dict): Object containing vote target info
  • pk (str): Object partition key
  • sk (str): Object sort key
  • slug (str): Object slug
  • entity (str): Entity type (e.g., "post", "comment")
• vote (str): Vote type - "up" or "down"

Returns: dict - Vote status

Example:

voted = API.run_job('vote.votecast', {
    "parent": {
        "pk": "post#abc123",
        "sk": "post#abc123",
        "slug": "my-post-slug",
        "entity": "post"
    },
    "vote": "up"
})

---

Reaction Jobs

Manage reactions to posts, comments, and other content.

reaction.reacttosomething()

React to something (post, comment, etc.).

reaction = API.run_job('reaction.reacttosomething', payload)

Parameters:

• parent (dict): Object containing pk and sk keys
  • pk (str): Object partition key
  • sk (str): Object sort key
• reaction (str): Reaction type (e.g., "like", "heart", "fire")

Returns: dict - Reaction status

Example:

reacted = API.run_job('reaction.reacttosomething', {
    "parent": {
        "pk": "post-id",
        "sk": "post-id"
    },
    "reaction": "like"
})

---

reaction.getbyparent()

Get a specific reaction by parent sort key.

reactions = API.run_job('reaction.getbyparent', parent_sk)

Parameters:

• parent_sk (str): Parent sort key (not full ID)

Returns: dict - Reaction data

Example:

reaction = API.run_job('reaction.getbyparent', "post#my-post-slug")

---

reaction.listbyparent()

List all reactions for a parent object.

reactions = API.run_job('reaction.listbyparent', parent_sk)

Parameters:

• parent_sk (str): Parent sort key (not full ID)

Returns: dict - List of reactions

Example:

reactions = API.run_job('reaction.listbyparent', "post#my-post-slug")

---

Notification Jobs

Manage user notifications.

notification.listnotifications()

Lists all notifications.

notifications = API.run_job('notification.listnotifications')

Parameters: None

Returns: dict - List of notifications

Example:

notifications = API.run_job('notification.listnotifications')
for notif in notifications.get('Items', []):
    print(f"{notif.get('notificationType')}: {notif.get('text')}")

---

notification.markallasread()

Marks all notifications as read.

API.run_job('notification.markallasread')

Parameters: None

Returns: dict - Mark as read status

Example:

API.run_job('notification.markallasread')

---

Trust Jobs

Manage trust relationships.

trust.createorupdate()

Create or update a trust relationship.

trust = API.run_job('trust.createorupdate', payload)

Parameters:

• parent (dict): Object containing pk and sk for the userprofile
• parentSlug (str): Slug of the userprofile
• trustLevel (str): Trust level (e.g., "trusted", "neutral", "distrusted")

Returns: dict - Trust relationship status

Example:

trust = API.run_job('trust.createorupdate', {
    "parent": {
        "pk": "userprofile#philosopher-jon",
        "sk": "userprofile#philosopher-jon"
    },
    "parentSlug": "philosopher-jon",
    "trustLevel": "trusted"
})

---

trust.listbyusersinit()

Lists trust relationships initialized by a user.

trusts = API.run_job('trust.listbyusersinit', user_id)

Parameters:

• user_id (str): The user ID (not username) whose trust relationships to list
• lastEvaluatedKey (dict, optional): Pagination key

Returns: dict - List of trust relationships

Example:

my_trusts = API.run_job('trust.listbyusersinit', "user-id-123")

---

trust.listbyuserhas()

Lists trust relationships where a user was the target.

trusts = API.run_job('trust.listbyuserhas', user_id)

Parameters:

• user_id (str): The user ID (not username) who was targeted

Returns: dict - List of trust relationships

---

Branch Jobs

Manage branches (platform branches or sub-wikis).

branch.get()

Get a specific branch.

branch = API.run_job('branch.get', branch_slug)

Parameters:

• branch_slug (str): The branch/subwiki slug identifier

Returns: dict - Branch data

Example:

branch = API.run_job('branch.get', "music")
print(branch.get('slug'))

---

branch.listbyname()

List all branches with pagination.

branches = API.run_job('branch.listbyname', lastEvaluatedKey)

Parameters:

• lastEvaluatedKey (dict, optional): Pagination key for next page

Returns: dict - List of branches

Note: Despite the name, this does not filter by branch name. It lists all branches.

Example:

# Get first page of branches
branches = API.run_job('branch.listbyname')
for branch in branches.get('Items', []):
    print(branch.get('slug'))

# Get next page
if 'LastEvaluatedKey' in branches:
    next_page = API.run_job('branch.listbyname', branches['LastEvaluatedKey'])

---

Feed Jobs

Get content feeds from TrustCafé.

feed.cafefeed()

Get the café-wide feed.

feed = API.run_job('feed.cafefeed')

Parameters:

• lastEvaluatedKey (dict, optional): Pagination key

Returns: dict - Café feed content

Example:

cafefeed = API.run_job('feed.cafefeed')
print(f"Feed items: {len(cafefeed.get('Items', []))}")

---

feed.followingfeed()

Get feed for users you follow.

feed = API.run_job('feed.followingfeed')

Parameters:

• lastEvaluatedKey (dict, optional): Pagination key

Returns: dict - Following feed content

Example:

following_feed = API.run_job('feed.followingfeed')
print(f"Following feed items: {len(following_feed.get('Items', []))}")

---

Block Jobs

⚠️ NOT IMPLEMENTED

Block functionality is not yet implemented in this wrapper. The TrustCafé API may support blocking, but no job or wrapper functions exist yet.

Status: TODO - See development.md for details

---

Mute Jobs

⚠️ NOT IMPLEMENTED

Mute functionality is not yet implemented in this wrapper. The TrustCafé API may support muting, but no job or wrapper functions exist yet.

Status: TODO - See development.md for details

---

Error Handling

Common Error Cases

Authentication Errors

try:
    profile = API.run_job('userprofile.get', "username")
except Exception as e:
    print(f"Authentication required: {e}")
    API.handle_token()
    profile = API.run_job('userprofile.get', "username")

Invalid Parameters

try:
    post = API.run_job('post.create', {
        "postText": "Test",
        # Missing required 'parent' field
    })
except Exception as e:
    print(f"Validation error: {e}")

Network Errors

try:
    result = API.make_request("GET", "content", "some/path")
except Exception as e:
    print(f"Network error: {e}")
    # Wait and retry
    import time
    time.sleep(1)
    result = API.make_request("GET", "content", "some/path")

Error Response Structure

API errors typically return:

{
    "error": "string",
    "message": "string",
    "details": {}  // Optional details
}

---

Response Structure

Most API responses follow a similar structure:

{
    "Items": [  // For list operations
        { /* item data */ }
    ],
    "Count": int,  // Number of items
    "LastEvaluatedKey": { /* pagination key */ },  // For pagination
    "ResponseMetadata": { /* AWS SDK metadata */ }
}

Pagination

For large result sets, results may include pagination:

def get_all_posts():
    posts = []
    last_key = None

    while True:
        if last_key:
            response = API.run_job('post.listpublic', last_key)
        else:
            response = API.run_job('post.listpublic')
        posts.extend(response.get('Items', []))

        has_more = 'LastEvaluatedKey' in response
        if not has_more:
            break

        last_key = response['LastEvaluatedKey']

    return posts

---

Best Practices

1. Use Type Hints

# Good
from typing import Optional, Dict, Any

def get_post(api: trustcafeapiwrapper.APIClient, post_id: str) -> Dict[str, Any]:
    return api.run_job('post.get', post_id)

# Bad
def get_post(api, post_id):
    return api.run_job('post.get', post_id)

2. Validate Responses

post = API.run_job('post.get', "post-id")
if not post or 'Items' not in post:
    raise ValueError("Invalid post data received")

item = post['Items'][0]
if not item:
    raise ValueError("Post item is empty")

3. Error Logging

import logging

logger = logging.getLogger(__name__)

try:
    profile = API.run_job('userprofile.get', username)
    logger.info(f"Successfully retrieved profile for {username}")
except Exception as e:
    logger.error(f"Failed to get profile for {username}: {e}")
    raise

4. Timeout Configurations

The API client has a default timeout of 20 seconds:

# The make_request method accepts a timeout parameter
# However, current implementation doesn't expose it directly
# Consider updating the APIClient to accept custom timeouts

5. Rate Limiting

Implement rate limiting for production use (see main README for example).

---

Next Steps

• Wrappers Guide - High-level wrappers for common tasks
• Custom Requests Guide - Making advanced API calls
• Troubleshooting Guide - Common issues and solutions