Python Examples

These examples demonstrate how to use the Tailwind API with Python.

Setup

pip install requests

import os
import requests
from datetime import datetime, timedelta

API_URL = 'https://api-v1.tailwind.ai'
API_KEY = os.environ['TAILWIND_API_KEY']

headers = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json'
}

List Accounts

def list_accounts():
    response = requests.get(
        f'{API_URL}/v1/accounts',
        headers=headers
    )
    response.raise_for_status()
    return response.json()['data']['accounts']

# Usage
accounts = list_accounts()
for account in accounts:
    print(f"Account: {account['displayName']} ({account['id']})")

List Boards

def list_boards(account_id: str):
    response = requests.get(
        f'{API_URL}/v1/accounts/{account_id}/boards',
        headers=headers
    )
    response.raise_for_status()
    return response.json()['data']['boards']

# Usage
boards = list_boards('acc_123456')
for board in boards:
    print(f"Board: {board['name']} ({board['id']})")

Create and Schedule a Post

def create_post(account_id: str, post_data: dict):
    response = requests.post(
        f'{API_URL}/v1/accounts/{account_id}/posts',
        headers=headers,
        json=post_data
    )
    response.raise_for_status()
    return response.json()['data']['post']

# Usage - Create a scheduled post
post = create_post('acc_123456', {
    'mediaUrl': 'https://example.com/my-image.jpg',
    'title': 'Amazing Recipe',
    'description': 'Try this delicious recipe! #recipes #cooking',
    'url': 'https://myblog.com/recipe',
    'boardId': 'board_789',
    'altText': 'A colorful plate of food',
    'sendAt': '2024-01-20T14:00:00Z'
})

print(f"Created post: {post['id']}")
print(f"Status: {post['status']}")

Create a Draft Post

# Omit sendAt to create a draft
draft = create_post('acc_123456', {
    'mediaUrl': 'https://example.com/my-image.jpg',
    'title': 'Draft for Review',
    'description': 'I will schedule this later',
    'boardId': 'board_789'
})

print(f"Draft created: {draft['id']}")
# draft['status'] will be 'draft'

Schedule a Draft

def schedule_post(account_id: str, post_id: str, send_at: str):
    response = requests.post(
        f'{API_URL}/v1/accounts/{account_id}/posts/{post_id}/schedule',
        headers=headers,
        json={'sendAt': send_at}
    )
    response.raise_for_status()
    return response.json()['data']['post']

# Usage
scheduled = schedule_post(
    'acc_123456',
    'post_abc123',
    '2024-01-25T09:00:00Z'
)

print(f"Scheduled for: {datetime.fromtimestamp(scheduled['sendAt'])}")

List Posts with Filtering

def list_posts(account_id: str, **kwargs):
    params = {}
    if 'status' in kwargs:
        params['status'] = kwargs['status']
    if 'limit' in kwargs:
        params['limit'] = kwargs['limit']
    if 'cursor' in kwargs:
        params['cursor'] = kwargs['cursor']
    if 'start_date' in kwargs:
        params['startDate'] = kwargs['start_date']
    if 'end_date' in kwargs:
        params['endDate'] = kwargs['end_date']

    response = requests.get(
        f'{API_URL}/v1/accounts/{account_id}/posts',
        headers=headers,
        params=params
    )
    response.raise_for_status()
    data = response.json()['data']
    return {
        'posts': data['posts'],
        'cursor': data.get('cursor')
    }

# List queued posts (default)
queued = list_posts('acc_123456')
print(f"Queued posts: {len(queued['posts'])}")

# List drafts
drafts = list_posts('acc_123456', status='draft')
print(f"Drafts: {len(drafts['posts'])}")

# List sent posts in a date range
sent = list_posts(
    'acc_123456',
    status='sent',
    start_date='2024-01-01T00:00:00Z',
    end_date='2024-01-31T23:59:59Z',
    limit=100
)
print(f"Sent in January: {len(sent['posts'])}")

Delete a Post

def delete_post(account_id: str, post_id: str):
    response = requests.delete(
        f'{API_URL}/v1/accounts/{account_id}/posts/{post_id}',
        headers=headers
    )
    response.raise_for_status()
    return True

# Usage
delete_post('acc_123456', 'post_abc123')
print('Post deleted')

Complete Example: Bulk Scheduler

import time

def bulk_schedule(account_id: str, posts: list):
    """Schedule multiple posts with a delay between each."""
    results = []

    for post_data in posts:
        try:
            post = create_post(account_id, post_data)
            results.append({'success': True, 'post': post})
            print(f"Scheduled: {post['id']}")

            # Wait 200ms between requests to avoid rate limits
            time.sleep(0.2)
        except Exception as e:
            results.append({
                'success': False,
                'error': str(e),
                'data': post_data
            })
            print(f"Failed: {e}")

    return results

# Usage
posts_to_schedule = [
    {
        'mediaUrl': 'https://example.com/image1.jpg',
        'title': 'Post 1',
        'description': 'First post',
        'boardId': 'board_789',
        'sendAt': '2024-01-20T09:00:00Z'
    },
    {
        'mediaUrl': 'https://example.com/image2.jpg',
        'title': 'Post 2',
        'description': 'Second post',
        'boardId': 'board_789',
        'sendAt': '2024-01-20T14:00:00Z'
    },
    {
        'mediaUrl': 'https://example.com/image3.jpg',
        'title': 'Post 3',
        'description': 'Third post',
        'boardId': 'board_789',
        'sendAt': '2024-01-20T19:00:00Z'
    }
]

results = bulk_schedule('acc_123456', posts_to_schedule)
successful = sum(1 for r in results if r['success'])
print(f"Scheduled {successful} posts")

Using Dataclasses for Type Safety

from dataclasses import dataclass
from typing import Optional, List
from datetime import datetime

@dataclass
class Account:
    id: str
    user_id: str
    display_name: str
    username: str
    avatar_url: Optional[str]
    token_authorized: bool
    is_domain_verified: bool
    created_at: str

@dataclass
class Board:
    id: str
    name: str
    is_collaborator: bool
    is_secret: bool

@dataclass
class Post:
    id: str
    status: str  # 'draft' | 'queued' | 'sent' | 'uploading'
    media_url: str
    media_type: str  # 'image' | 'video'
    title: Optional[str]
    description: Optional[str]
    url: Optional[str]
    board_id: Optional[str]
    alt_text: Optional[str]
    send_at: Optional[int]
    sent_at: Optional[int]
    created_at: int
    pin_id: Optional[str]

def parse_account(data: dict) -> Account:
    return Account(
        id=data['id'],
        user_id=data['userId'],
        display_name=data['displayName'],
        username=data['username'],
        avatar_url=data.get('avatarUrl'),
        token_authorized=data['tokenAuthorized'],
        is_domain_verified=data['isDomainVerified'],
        created_at=data['createdAt']
    )

def parse_post(data: dict) -> Post:
    return Post(
        id=data['id'],
        status=data['status'],
        media_url=data['mediaUrl'],
        media_type=data['mediaType'],
        title=data.get('title'),
        description=data.get('description'),
        url=data.get('url'),
        board_id=data.get('boardId'),
        alt_text=data.get('altText'),
        send_at=data.get('sendAt'),
        sent_at=data.get('sentAt'),
        created_at=data['createdAt'],
        pin_id=data.get('pinId')
    )

Error Handling

from requests.exceptions import HTTPError

def safe_api_call(func):
    """Decorator for handling API errors."""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except HTTPError as e:
            if e.response.status_code == 401:
                print('Authentication failed. Check your API key.')
            elif e.response.status_code == 429:
                print('Rate limit exceeded. Please wait before retrying.')
            elif e.response.status_code == 404:
                print('Resource not found. Check your IDs.')
            else:
                print(f'API error: {e}')
            raise
    return wrapper

@safe_api_call
def get_accounts():
    return list_accounts()

# Usage
accounts = get_accounts()

Async Example with aiohttp

import asyncio
import aiohttp

async def list_accounts_async():
    async with aiohttp.ClientSession() as session:
        async with session.get(
            f'{API_URL}/v1/accounts',
            headers=headers
        ) as response:
            data = await response.json()
            return data['data']['accounts']

async def create_posts_async(account_id: str, posts: list):
    """Create multiple posts concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for post_data in posts:
            task = session.post(
                f'{API_URL}/v1/accounts/{account_id}/posts',
                headers=headers,
                json=post_data
            )
            tasks.append(task)

        responses = await asyncio.gather(*tasks, return_exceptions=True)
        results = []
        for i, response in enumerate(responses):
            if isinstance(response, Exception):
                results.append({'success': False, 'error': str(response)})
            else:
                data = await response.json()
                results.append({'success': True, 'post': data['data']['post']})

        return results

# Usage
accounts = asyncio.run(list_accounts_async())