> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/trailbaseio/trailbase/llms.txt
> Use this file to discover all available pages before exploring further.

# 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:

```bash theme={null}
pip install trailbase
```

Or using Poetry:

```bash theme={null}
poetry add trailbase
```

## Requirements

* Python 3.13+
* httpx
* PyJWT
* cryptography

## Initialization

### Basic Client

```python theme={null}
from trailbase import Client

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

### Client with Tokens

```python theme={null}
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

```python theme={null}
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

```python theme={null}
try:
    tokens = client.login('user@example.com', '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

```python theme={null}
client.logout()
```

### Current User

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

### Access Tokens

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
posts.update('post-id', {
    'title': 'Updated Title'
})
```

### Delete a Record

```python theme={null}
posts.delete('post-id')
```

## Filtering

```python theme={null}
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

```python theme={null}
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):

```python theme={null}
# 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

```python theme={null}
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

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

data = response.json()
```

### Streaming Responses

```python theme={null}
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

```python theme={null}
class User:
    id: str
    email: str
```

### Tokens

```python theme={null}
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

```python theme={null}
class ListResponse:
    cursor: str | None
    total_count: int | None
    records: list[dict[str, Any]]
```

### RecordId

```python theme={null}
class RecordId:
    id: str
    
    def __repr__(self) -> str:
        return self.id
```

## Async Support

<Note>
  The current Python SDK is synchronous. For async operations, use `httpx.AsyncClient` and adapt the code accordingly.
</Note>

## Best Practices

<Warning>
  Store tokens securely and never commit them to version control.
</Warning>

<Info>
  The client automatically refreshes auth tokens before they expire.
</Info>

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

## Example Application

```python theme={null}
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()
```