Skip to main content

Python SDK

Official Python SDK for the Flow API.

Installation​

pip install flow-sdk

Quick Start​

from flow_sdk import Flow
from datetime import datetime

flow = Flow(api_key='flow_sk_live_abc12345_xyz789...')

# Create a post
post = flow.posts.create(
    channel_id='channel_123',
    content='Hello from Flow!',
    scheduled_for=datetime(2024, 12, 25, 10, 0, 0)
)

print(f'Post created: {post.id}')

Configuration​

from flow_sdk import Flow

flow = Flow(
    api_key='flow_sk_live_...',
    base_url='https://api.flow.dev',  # default
    timeout=30,  # seconds, default
    max_retries=3,  # default
)

Posts​

List Posts​

# List all posts
posts = flow.posts.list()

# With filters
posts = flow.posts.list(
    filter={'status': 'queued', 'channelId': 'channel_123'},
    sort='-scheduledFor',
    page=1,
    per_page=50
)

Create Post​

# Immediate post
post = flow.posts.create(
    channel_id='channel_123',
    content='Hello!'
)

# Scheduled post
post = flow.posts.create(
    channel_id='channel_123',
    content='Scheduled post',
    scheduled_for=datetime(2024, 12, 25, 10, 0, 0),
    media_keys=['media_key_123'],
    enable_optimization=True
)

Get Post​

post = flow.posts.get('post_123')

Update Post​

flow.posts.update('post_123', {
    'content': 'Updated content',
    'scheduled_for': datetime(2024, 12, 26, 10, 0, 0)
})

Delete Post​

flow.posts.delete('post_123')

Batch Operations​

# Batch create
result = flow.posts.create_batch({
    'posts': [
        {'channel_id': 'channel_123', 'content': 'Post 1'},
        {'channel_id': 'channel_123', 'content': 'Post 2'},
    ]
})

# Batch update
flow.posts.update_batch({
    'posts': [
        {'id': 'post_1', 'content': 'Updated 1'},
        {'id': 'post_2', 'content': 'Updated 2'},
    ]
})

# Batch delete
flow.posts.delete_batch({
    'post_ids': ['post_1', 'post_2', 'post_3']
})

Channels​

List Channels​

channels = flow.channels.list()

Create Channel​

channel = flow.channels.create(
    name='My Channel',
    color='#3B82F6'
)

Get Channel​

channel = flow.channels.get('channel_123')

Update Channel​

flow.channels.update('channel_123', {
    'name': 'Updated Name',
    'color': '#FF0000'
})

Delete Channel​

flow.channels.delete('channel_123')

Webhooks​

List Webhooks​

webhooks = flow.webhooks.list()

Create Webhook​

webhook = flow.webhooks.create(
    url='https://example.com/webhooks/flow',
    events=['post.delivered', 'post.failed']
)

print(f'Webhook secret: {webhook.secret}')  # Save this!

Get Webhook​

webhook = flow.webhooks.get('webhook_123')

Update Webhook​

flow.webhooks.update('webhook_123', {
    'url': 'https://new-url.com/webhooks/flow',
    'events': ['post.delivered'],
    'active': False
})

Delete Webhook​

flow.webhooks.delete('webhook_123')

Get Deliveries​

deliveries = flow.webhooks.get_deliveries('webhook_123', {
    'status': 'failed',
    'page': 1,
    'per_page': 50
})

Test Webhook​

flow.webhooks.test('webhook_123')

API Keys​

List API Keys​

keys = flow.api_keys.list()

Create API Key​

result = flow.api_keys.create({
    'name': 'Production Key',
    'permissions': ['posts:write', 'channels:read']
})

print(f'API Key: {result.api_key}')  # Save this!

Get API Key​

key = flow.api_keys.get('key_123')

Update API Key​

flow.api_keys.update('key_123', {
    'name': 'Updated Name'
})

Delete API Key​

flow.api_keys.delete('key_123')

Media​

Upload Media​

with open('image.jpg', 'rb') as f:
    media = flow.media.upload(
        file=f.read(),
        filename='image.jpg'
    )

print(f'Media key: {media.key}')

Get Media URL​

url = flow.media.get_url('media_key_123')

Delete Media​

flow.media.delete('media_key_123')

Error Handling​

from flow_sdk import Flow, FlowError

try:
    post = flow.posts.create(
        channel_id='invalid',
        content='Test'
    )
except FlowError as e:
    if e.status == 401:
        print('Authentication failed')
    elif e.status == 403:
        print('Permission denied')
    elif e.status == 404:
        print('Resource not found')
    elif e.status == 429:
        print(f'Rate limited. Retry after {e.retry_after}s')
    else:
        print(f'API error: {e.message}')
    
    # Access error details
    if e.details:
        for detail in e.details:
            print(f"{detail['field']}: {detail['message']}")

Examples​

Schedule Weekly Posts​

from datetime import datetime, timedelta

def schedule_weekly_posts(channel_id: str, contents: list):
    posts = []
    for i, content in enumerate(contents):
        date = datetime.now() + timedelta(days=i+1)
        date = date.replace(hour=9, minute=0, second=0, microsecond=0)
        
        posts.append({
            'channel_id': channel_id,
            'content': content,
            'scheduled_for': date
        })
    
    return flow.posts.create_batch({'posts': posts})

# Usage
schedule_weekly_posts('channel_123', [
    'Monday motivation!',
    'Tuesday tips',
    'Wednesday wisdom',
])

Monitor Post Status​

import time

def wait_for_post_delivery(post_id: str):
    while True:
        post = flow.posts.get(post_id)
        
        if post.status == 'done':
            print('Post delivered!')
            break
        elif post.status == 'blocked':
            print('Post blocked')
            break
        
        time.sleep(5)  # Wait 5 seconds

Next Steps​