Python SDK

Official Python client for TrailBase

The official Python SDK for TrailBase provides a fully-typed client for accessing your TrailBase backend from Python applications.

Installation

Install using pip:

pip install trailbase

Or using Poetry:

poetry add trailbase

Requirements

• Python 3.13+
• httpx
• PyJWT
• cryptography

Initialization

Basic Client

from trailbase import Client

client = Client('https://your-server.trailbase.io')

Client with Tokens

from trailbase import Client, Tokens

tokens = Tokens(
    auth='your-auth-token',
    refresh='your-refresh-token',
    csrf='your-csrf-token'
)

client = Client('https://your-server.trailbase.io', tokens=tokens)

Custom HTTP Client

import httpx
from trailbase import Client

http_client = httpx.Client(timeout=30.0)
client = Client('https://your-server.trailbase.io', http_client=http_client)

Authentication

Login

try:
    tokens = client.login('[e-mail]', 'password')
    print(f'Auth token: {tokens.auth}')
    
    user = client.user()
    if user:
        print(f'Logged in as: {user.email}')
except Exception as e:
    print(f'Login failed: {e}')

Logout

client.logout()

Current User

user = client.user()
if user:
    print(f'User ID: {user.id}')
    print(f'Email: {user.email}')

Access Tokens

tokens = client.tokens()
if tokens:
    # Persist tokens for later use
    import json
    with open('tokens.json', 'w') as f:
        json.dump(tokens.to_json(), f)

Record API

List Records

from trailbase import Client

client = Client('https://your-server.trailbase.io')
posts = client.records('posts')

response = posts.list(
    limit=10,
    offset=0,
    order=['-created_at'],
    count=True
)

print(f'Records: {response.records}')
print(f'Total count: {response.total_count}')
print(f'Next cursor: {response.cursor}')

Read a Record

# Using string ID
post = posts.read('post-id')

# Using integer ID
post = posts.read(123)

# With RecordId type
from trailbase import RecordId
post_id = RecordId('post-id')
post = posts.read(post_id)

# With expanded relationships
post_with_author = posts.read('post-id', expand=['author'])

print(f"Title: {post['title']}")

Create a Record

new_post = {
    'title': 'Hello World',
    'content': 'My first post from Python'
}

post_id = posts.create(new_post)
print(f'Created post with ID: {post_id}')

Create Multiple Records

new_posts = [
    {'title': 'Post 1', 'content': 'Content 1'},
    {'title': 'Post 2', 'content': 'Content 2'}
]

ids = posts.create_bulk(new_posts)
print(f'Created {len(ids)} posts')

Update a Record

posts.update('post-id', {
    'title': 'Updated Title'
})

Delete a Record

posts.delete('post-id')

Filtering

from trailbase import Filter, CompareOp, And, Or

# Simple equality filter
response = posts.list(
    filters=[
        Filter(column='author_id', value=user_id)
    ]
)

# With comparison operators
from datetime import datetime, timedelta

week_ago = int((datetime.now() - timedelta(days=7)).timestamp())
recent_posts = posts.list(
    filters=[
        Filter(
            column='created_at',
            op=CompareOp.GREATER_THAN,
            value=str(week_ago)
        )
    ]
)

# LIKE operator for text search
search_results = posts.list(
    filters=[
        Filter(
            column='title',
            op=CompareOp.LIKE,
            value='%search%'
        )
    ]
)

# AND composite filter
filtered = posts.list(
    filters=[
        And([
            Filter(column='status', value='published'),
            Filter(column='author_id', value=user_id)
        ])
    ]
)

# OR composite filter
filtered = posts.list(
    filters=[
        Or([
            Filter(column='category', value='tech'),
            Filter(column='category', value='science')
        ])
    ]
)

Available Comparison Operators

from trailbase import CompareOp

# Available operators:
CompareOp.EQUAL
CompareOp.NOT_EQUAL
CompareOp.LESS_THAN
CompareOp.LESS_THAN_EQUAL
CompareOp.GREATER_THAN
CompareOp.GREATER_THAN_EQUAL
CompareOp.LIKE
CompareOp.REGEXP
CompareOp.ST_WITHIN       # Geospatial
CompareOp.ST_INTERSECTS   # Geospatial
CompareOp.ST_CONTAINS     # Geospatial

Real-time Subscriptions

Subscribe to record changes using Server-Sent Events (SSE):

# Subscribe to a single record
for event in posts.subscribe('post-id'):
    if 'Insert' in event:
        print(f'Record inserted: {event["Insert"]}')
    elif 'Update' in event:
        print(f'Record updated: {event["Update"]}')
    elif 'Delete' in event:
        print(f'Record deleted: {event["Delete"]}')
    elif 'Error' in event:
        print(f'Error: {event["Error"]}')

Error Handling

try:
    post = posts.read('post-id')
except Exception as e:
    print(f'Error: {e}')
    if hasattr(e, 'status_code'):
        print(f'Status code: {e.status_code}')

Advanced Usage

Custom Fetch

response = client.fetch(
    'api/custom/endpoint',
    method='POST',
    data={'key': 'value'}
)

data = response.json()

Streaming Responses

import httpx

with client.stream(
    'api/records/v1/posts/subscribe/123',
    timeout=httpx.Timeout(None)
) as response:
    if response.status_code == 200:
        for line in response.iter_lines():
            if line.startswith('data: '):
                import json
                event = json.loads(line[6:])
                print(event)

Type Definitions

User

class User:
    id: str
    email: str

Tokens

class Tokens:
    auth: str
    refresh: str | None
    csrf: str | None
    
    def to_json(self) -> dict[str, str | None]:
        return {
            'auth_token': self.auth,
            'refresh_token': self.refresh,
            'csrf_token': self.csrf
        }

ListResponse

class ListResponse:
    cursor: str | None
    total_count: int | None
    records: list[dict[str, Any]]

RecordId

class RecordId:
    id: str
    
    def __repr__(self) -> str:
        return self.id

Async Support

The current Python SDK is synchronous. For async operations, use httpx.AsyncClient and adapt the code accordingly.

Best Practices

Store tokens securely and never commit them to version control.

The client automatically refreshes auth tokens before they expire.

Use context managers or try-finally blocks to ensure proper cleanup of HTTP connections.

Example Application

from trailbase import Client, Filter, CompareOp
import os

def main():
    # Initialize client
    client = Client(os.getenv('TRAILBASE_URL', 'http://localhost:4000'))
    
    # Login
    try:
        client.login(
            os.getenv('TRAILBASE_EMAIL'),
            os.getenv('TRAILBASE_PASSWORD')
        )
        print(f'Logged in as: {client.user().email}')
    except Exception as e:
        print(f'Login failed: {e}')
        return
    
    # List posts
    posts = client.records('posts')
    response = posts.list(
        order=['-created_at'],
        limit=10,
        filters=[
            Filter(
                column='published',
                value='true'
            )
        ]
    )
    
    print(f'\nFound {len(response.records)} posts:')
    for post in response.records:
        print(f"- {post['title']}")
    
    # Create a new post
    new_post_id = posts.create({
        'title': 'Hello from Python',
        'content': 'This post was created using the TrailBase Python SDK',
        'published': True
    })
    print(f'\nCreated new post with ID: {new_post_id}')
    
    # Read the post
    post = posts.read(new_post_id)
    print(f'Post title: {post["title"]}')
    
    # Logout
    client.logout()
    print('\nLogged out')

if __name__ == '__main__':
    main()