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"),
    env="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: None

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.

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

Parameters: None

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_name)

Parameters:

• branchName (str): The branch identifier

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', username)

Parameters:

• username (str): The username of the user

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', post_id)

Parameters:

• postId (str): The post identifier

Returns: dict - Removed post data

Example:

removed_post = API.run_job('post.listremoved', "deleted-post-id")

---

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:

• postId (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()

Follows a user.

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

Parameters:

• username (str): The username to follow

Returns: dict - Follow status

Example:

followed = API.run_job('follow.follow', "another-user")
print(f"Follow status: {followed}")

---

Vote Jobs

Cast votes on posts/content.

vote.votecast()

Casts a vote on a post.

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

Parameters:

• pk (str): Object partition key
• sk (str): Object sort key
• voteType (str): Vote type - "up" or "down"

Returns: dict - Vote status

Example:

voted = API.run_job('vote.votecast', {
    "pk": "post-id",
    "sk": "post-id",
    "voteType": "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:

• objectPK (str): Object partition key
• objectSK (str): Object sort key
• objectType (str): Type of object ("post", "comment", etc.)
• reactionType (str): Reaction type (depends on API)

Returns: dict - Reaction status

Example:

reacted = API.run_job('reaction.reacttosomething', {
    "objectPK": "post-id",
    "objectSK": "post-id",
    "objectType": "post",
    "reactionType": "like"
})

---

reaction.getbyparent()

Get reactions by parent object.

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

Parameters:

• objectId (str): Parent object ID

Returns: dict - List of reactions

---

reaction.listbyparent()

List all reactions for a parent object.

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

Parameters:

• parentPK (str): Parent partition key
• parentSK (str): Parent sort key

Returns: dict - List of reactions

---

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:

• pk (str): Trust identifier (partition key)
• sk (str): Trust identifier (sort key)
• trustType (str): Trust type ("positive", "negative", etc.)
• Other optional fields

Returns: dict - Trust relationship status

Example:

trust = API.run_job('trust.createorupdate', {
    "pk": "trust-id-123",
    "sk": "trust-id-123",
    "trustType": "positive"
})

---

trust.listbyusersinit()

Lists trust relationships initialized by a user.

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

Parameters:

• username (str): The user whose trust relationships to list

Returns: dict - List of trust relationships

Example:

my_trusts = API.run_job('trust.listbyusersinit', "your-username")

---

trust.listbyuserhas()

Lists trust relationships where a user was the target.

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

Parameters:

• username (str): The user 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_name)

Parameters:

• branchName (str): The branch identifier

Returns: dict - Branch data

Example:

branch = API.run_job('branch.get', "main")
print(branch.get('branchName'))

---

branch.listbyname()

List branches by name prefix.

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

Parameters:

• prefix (str): String prefix for branch names

Returns: dict - List of matching branches

Example:

music_branches = API.run_job('branch.listbyname', "music")

---

Feed Jobs

Get content feeds from TrustCafé.

feed.cafefeed()

Get the café-wide feed.

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

Parameters: None

Returns: dict - Café feed content

Example:

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

---

feed.following()

Get feed for users you follow.

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

Parameters: None

Returns: dict - Following feed content

Example:

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

---

Block Jobs

Block users or content.

block.[notimplemented]

Block functionality is defined in the backend but not yet implemented in the wrapper.

Status: TODO

Implementation needed:

• Job function creation
• Payload structure definition
• Error handling

---

Mute Jobs

Mute users or content.

mute.[notimplemented]

Mute functionality is pending implementation.

Status: TODO

Implementation needed:

• Job function creation
• Payload structure definition
• Error handling

---

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 = []
    paginator = None

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

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

        # Set params for next page
        params = {
            'ExclusiveStartKey': 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