Python SDK

pip install trailbase

poetry add trailbase

from trailbase import Client

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

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)

import httpx
from trailbase import Client

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

try:
    tokens = client.login('[email protected]', '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}')

client.logout()

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

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)

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}')

# 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']}")

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}')

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

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

posts.delete('post-id')

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')
        ])
    ]
)

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

# 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"]}')

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}')

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

data = response.json()

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)

class User:
    id: str
    email: str

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
        }

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

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

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