Stack0 Python SDK

from stack0 import Stack0
from stack0.mail.types import SendEmailRequest

stack0 = Stack0(api_key="stack0_...")

# Send an email
response = stack0.mail.send(SendEmailRequest(
    from_="[email protected]",
    to="[email protected]",
    subject="Hello from Stack0",
    html="<h1>Welcome</h1><p>Thanks for signing up.</p>",
))
print(response.id, response.status)

stack0.close()

import asyncio
from stack0 import AsyncStack0
from stack0.mail.types import SendEmailRequest

async def main():
    async with AsyncStack0(api_key="stack0_...") as stack0:
        response = await stack0.mail.send(SendEmailRequest(
            from_="[email protected]",
            to="[email protected]",
            subject="Hello from Stack0",
            html="<h1>Welcome</h1>",
        ))
        print(response.id)

asyncio.run(main())

from stack0 import Stack0

with Stack0(api_key="stack0_...") as stack0:
    analytics = stack0.mail.get_analytics()
    print(f"Delivery rate: {analytics.delivery_rate}%")

from stack0 import Stack0, AsyncStack0

client = Stack0(
    api_key="stack0_...",       # Required. Your Stack0 API key.
    base_url=None,              # Optional. Custom API base URL.
    cdn_url=None,               # Optional. CDN base URL for image transforms.
    timeout=30.0,               # Optional. Request timeout in seconds.
)

from stack0.mail.types import SendEmailRequest, SendBatchEmailRequest, SendBroadcastEmailRequest

# Single email
response = stack0.mail.send(SendEmailRequest(
    from_="[email protected]",
    to="[email protected]",
    subject="Order Confirmed",
    html="<p>Your order #1234 has been confirmed.</p>",
    tags=["transactional", "orders"],
))

# Batch (up to 100 emails)
batch = stack0.mail.send_batch(SendBatchEmailRequest(
    emails=[
        SendEmailRequest(from_="[email protected]", to="[email protected]", subject="Hi A", html="<p>Hello A</p>"),
        SendEmailRequest(from_="[email protected]", to="[email protected]", subject="Hi B", html="<p>Hello B</p>"),
    ],
))

# Broadcast (same content, up to 1000 recipients)
broadcast = stack0.mail.send_broadcast(SendBroadcastEmailRequest(
    from_="[email protected]",
    to=["[email protected]", "[email protected]"],
    subject="Weekly Update",
    html="<p>Here is your weekly update.</p>",
))

response = stack0.mail.send(SendEmailRequest(
    from_="[email protected]",
    to="[email protected]",
    subject="Welcome",
    template_id="tmpl_abc123",
    template_variables={"name": "Alice", "plan": "Pro"},
))

# Get email details
email = stack0.mail.get("email_id")

# List emails with filters
from stack0.mail.types import ListEmailsRequest
emails = stack0.mail.list(ListEmailsRequest(
    status="delivered",
    limit=50,
))

# Resend / cancel
stack0.mail.resend("email_id")
stack0.mail.cancel("email_id")

from stack0.mail.types import TimeSeriesAnalyticsRequest

analytics = stack0.mail.get_analytics()
print(f"Sent: {analytics.sent}, Delivery rate: {analytics.delivery_rate}")

timeseries = stack0.mail.get_time_series_analytics(TimeSeriesAnalyticsRequest(days=30))
hourly = stack0.mail.get_hourly_analytics()
senders = stack0.mail.list_senders()

from stack0.mail.types import ListDomainsRequest, AddDomainRequest

# List domains
domains = stack0.mail.domains.list(ListDomainsRequest(project_slug="my-project"))

# Add and verify a domain
result = stack0.mail.domains.add(AddDomainRequest(domain="mail.example.com"))
dns_records = stack0.mail.domains.get_dns_records("domain_id")
verification = stack0.mail.domains.verify("domain_id")

# Manage
stack0.mail.domains.set_default("domain_id")
stack0.mail.domains.delete("domain_id")

from stack0.mail.types import CreateTemplateRequest, PreviewTemplateRequest

template = stack0.mail.templates.create(CreateTemplateRequest(
    name="Welcome Email",
    slug="welcome",
    subject="Welcome, {{name}}!",
    html="<h1>Welcome, {{name}}</h1>",
))

# List, get, get by slug
templates = stack0.mail.templates.list()
template = stack0.mail.templates.get("template_id")
template = stack0.mail.templates.get_by_slug("welcome")

# Preview with variables
preview = stack0.mail.templates.preview(PreviewTemplateRequest(
    id="template_id",
    variables={"name": "Alice"},
))

from stack0.mail.types import CreateContactRequest, ImportContactsRequest, ImportContactInput

contact = stack0.mail.contacts.create(CreateContactRequest(
    email="[email protected]",
    first_name="Alice",
    last_name="Smith",
    metadata={"plan": "pro"},
))

# Bulk import
result = stack0.mail.contacts.import_contacts(ImportContactsRequest(
    contacts=[
        ImportContactInput(email="[email protected]", first_name="A"),
        ImportContactInput(email="[email protected]", first_name="B"),
    ],
    audience_id="aud_123",
))
print(f"Imported: {result.imported}, Skipped: {result.skipped}")

from stack0.mail.types import CreateAudienceRequest, AddContactsToAudienceRequest

audience = stack0.mail.audiences.create(CreateAudienceRequest(
    name="Newsletter Subscribers",
    description="Users who opted in to the newsletter",
))

stack0.mail.audiences.add_contacts(AddContactsToAudienceRequest(
    id=audience.id,
    contact_ids=["contact_1", "contact_2"],
))

from stack0.mail.types import CreateCampaignRequest, SendCampaignRequest

campaign = stack0.mail.campaigns.create(CreateCampaignRequest(
    name="Product Launch",
    subject="Introducing our new feature",
    from_email="[email protected]",
    html="<h1>Big news</h1>",
    audience_id="aud_123",
))

# Send immediately
result = stack0.mail.campaigns.send(SendCampaignRequest(
    id=campaign.id,
    send_now=True,
))

# Lifecycle
stack0.mail.campaigns.pause("campaign_id")
stack0.mail.campaigns.cancel("campaign_id")
stats = stack0.mail.campaigns.get_stats("campaign_id")

from stack0.mail.types import (
    CreateSequenceRequest, CreateNodeRequest, CreateConnectionRequest,
    SetNodeEmailRequest, SetNodeTimerRequest,
)

# Create a sequence
sequence = stack0.mail.sequences.create(CreateSequenceRequest(
    name="Onboarding Flow",
    trigger_type="manual",
))

# Add nodes
email_node = stack0.mail.sequences.create_node(CreateNodeRequest(
    id=sequence.id,
    node_type="email",
    name="Welcome Email",
    position_x=0,
    position_y=100,
))

timer_node = stack0.mail.sequences.create_node(CreateNodeRequest(
    id=sequence.id,
    node_type="timer",
    name="Wait 1 Day",
    position_x=0,
    position_y=200,
))

# Configure nodes
stack0.mail.sequences.set_node_email(sequence.id, SetNodeEmailRequest(
    node_id=email_node.id,
    subject="Welcome!",
    html="<p>Thanks for joining.</p>",
))

stack0.mail.sequences.set_node_timer(sequence.id, SetNodeTimerRequest(
    node_id=timer_node.id,
    delay_amount=1,
    delay_unit="days",
))

# Connect nodes
stack0.mail.sequences.create_connection(CreateConnectionRequest(
    id=sequence.id,
    source_node_id=email_node.id,
    target_node_id=timer_node.id,
))

# Publish
stack0.mail.sequences.publish(sequence.id)

from stack0.mail.types import CreateEventRequest, TrackEventRequest

event = stack0.mail.events.create(CreateEventRequest(
    name="user.signed_up",
    description="Fired when a user completes registration",
))

stack0.mail.events.track(TrackEventRequest(
    event_name="user.signed_up",
    contact_email="[email protected]",
    properties={"plan": "pro", "source": "landing_page"},
))

# Simple upload (handles presigned URL flow automatically)
with open("photo.jpg", "rb") as f:
    asset = stack0.cdn.upload(
        project_slug="my-project",
        file=f.read(),
        filename="photo.jpg",
        mime_type="image/jpeg",
        folder="uploads/photos",
    )
print(asset.url)

async with AsyncStack0(api_key="stack0_...") as stack0:
    with open("photo.jpg", "rb") as f:
        asset = await stack0.cdn.upload(
            project_slug="my-project",
            file=f.read(),
            filename="photo.jpg",
            mime_type="image/jpeg",
        )

from stack0.cdn.types import TransformOptions

url = stack0.cdn.get_transform_url(
    "https://cdn.example.com/assets/photo.jpg",
    TransformOptions(
        width=800,
        height=600,
        format="webp",
        quality=85,
    ),
)
# https://cdn.example.com/assets/photo.jpg?w=828&h=600&f=webp&q=85

from stack0.cdn.types import ListAssetsRequest, MoveAssetsRequest

# List assets
assets = stack0.cdn.list(ListAssetsRequest(
    project_slug="my-project",
    type="image",
    search="photo",
))

# Get, update, delete
asset = stack0.cdn.get("asset_id")
stack0.cdn.delete("asset_id")
stack0.cdn.delete_many(["asset_1", "asset_2"])

# Move assets between folders
stack0.cdn.move(MoveAssetsRequest(
    asset_ids=["asset_1", "asset_2"],
    target_folder="archive/2024",
))

from stack0.cdn.types import CreateFolderRequest, GetFolderTreeRequest

tree = stack0.cdn.get_folder_tree(GetFolderTreeRequest(project_slug="my-project"))
folder = stack0.cdn.create_folder(CreateFolderRequest(
    project_slug="my-project",
    name="photos",
    parent_id=None,
))
stack0.cdn.get_folder("folder_id")
stack0.cdn.get_folder_by_path("uploads/photos")
stack0.cdn.delete_folder("folder_id", delete_contents=True)

from stack0.cdn.types import TranscodeVideoRequest, GenerateGifRequest

# Transcode
job = stack0.cdn.transcode(TranscodeVideoRequest(
    asset_id="video_asset_id",
    project_slug="my-project",
))

# Check status
job = stack0.cdn.get_job(job.id)

# Get streaming URLs (HLS/DASH)
urls = stack0.cdn.get_streaming_urls("video_asset_id")

# Generate GIF from video segment
gif = stack0.cdn.generate_gif(GenerateGifRequest(
    asset_id="video_asset_id",
    project_slug="my-project",
    start_time=5.0,
    end_time=10.0,
    width=320,
))

# Upload private file
private_file = stack0.cdn.upload_private(
    project_slug="my-project",
    file=file_bytes,
    filename="contract.pdf",
    mime_type="application/pdf",
    description="Q4 contract",
)

# Generate time-limited download URL
from stack0.cdn.types import PrivateDownloadUrlRequest
download = stack0.cdn.get_private_download_url(PrivateDownloadUrlRequest(
    file_id=private_file.id,
    expires_in=3600,  # seconds
))
print(download.download_url)

from stack0.screenshots.types import CreateScreenshotRequest, GetScreenshotRequest

# Fire and forget
response = stack0.screenshots.capture(CreateScreenshotRequest(
    url="https://example.com",
    format="png",
    full_page=True,
    device_type="desktop",
))

# Capture and wait for result (polls until complete)
screenshot = stack0.screenshots.capture_and_wait(
    CreateScreenshotRequest(
        url="https://example.com",
        format="webp",
        full_page=True,
    ),
    poll_interval=1.0,   # seconds between polls
    timeout=60.0,        # max wait time
)
print(screenshot.image_url)

async with AsyncStack0(api_key="stack0_...") as stack0:
    screenshot = await stack0.screenshots.capture_and_wait(
        CreateScreenshotRequest(
            url="https://example.com",
            format="png",
            full_page=True,
        ),
    )

from stack0.screenshots.types import CreateBatchScreenshotsRequest

# Process multiple URLs at once
job = stack0.screenshots.batch_and_wait(
    CreateBatchScreenshotsRequest(
        urls=["https://example.com", "https://example.org"],
        format="png",
        full_page=True,
    ),
    poll_interval=2.0,
    timeout=300.0,
)

from stack0.screenshots.types import CreateScreenshotScheduleRequest

schedule = stack0.screenshots.create_schedule(CreateScreenshotScheduleRequest(
    url="https://example.com",
    frequency="daily",
    format="png",
    full_page=True,
))

# Manage schedules
stack0.screenshots.toggle_schedule(GetScheduleRequest(id=schedule.id))

from stack0.extraction.types import CreateExtractionRequest

# Markdown extraction
result = stack0.extraction.extract_and_wait(
    CreateExtractionRequest(
        url="https://example.com/blog/post",
        mode="markdown",
    ),
)
print(result.markdown)

# Schema-based extraction
result = stack0.extraction.extract_and_wait(
    CreateExtractionRequest(
        url="https://example.com/product",
        mode="schema",
        schema_={"name": "string", "price": "number", "in_stock": "boolean"},
        prompt="Extract the main product details",
    ),
)
print(result.extracted_data)

async with AsyncStack0(api_key="stack0_...") as stack0:
    result = await stack0.extraction.extract_and_wait(
        CreateExtractionRequest(
            url="https://example.com/article",
            mode="markdown",
            include_metadata=True,
        ),
    )
    print(result.page_metadata)

from stack0.extraction.types import CreateBatchExtractionsRequest

job = stack0.extraction.batch_and_wait(
    CreateBatchExtractionsRequest(
        urls=["https://example.com/page1", "https://example.com/page2"],
        mode="markdown",
    ),
)

usage = stack0.extraction.get_usage()
daily = stack0.extraction.get_usage_daily()

from stack0.integrations.types import InitiateOAuthRequest

# List existing connections
connections = stack0.integrations.list_connections()

# Start OAuth flow
oauth = stack0.integrations.initiate_oauth(InitiateOAuthRequest(
    connector_id="salesforce",
    redirect_url="https://app.example.com/callback",
))
print(oauth.authorization_url)

# Get connection details
connection = stack0.integrations.get_connection("conn_123")

# Delete a connection
stack0.integrations.delete_connection("conn_123")

from stack0.integrations.types import CreateContactInput

# Contacts
contacts = stack0.integrations.crm.list_contacts("conn_123")
contact = stack0.integrations.crm.create_contact("conn_123", CreateContactInput(
    email="[email protected]",
    first_name="Alice",
))

# Companies and Deals
companies = stack0.integrations.crm.list_companies("conn_123")
deals = stack0.integrations.crm.list_deals("conn_123")

files = stack0.integrations.storage.list_files("conn_123")
folders = stack0.integrations.storage.list_folders("conn_123")

from stack0.integrations.types import SendMessageInput

channels = stack0.integrations.communication.list_channels("conn_123")
stack0.integrations.communication.send_message("conn_123", SendMessageInput(
    channel_id="channel_123",
    content="Hello from Stack0!",
))

documents = stack0.integrations.productivity.list_documents("conn_123")
tables = stack0.integrations.productivity.list_tables("conn_123")

from stack0.marketing.types import DiscoverTrendsRequest

result = stack0.marketing.discover_trends(DiscoverTrendsRequest(
    project_slug="my-project",
    environment="production",
))

from stack0.marketing.types import CreateContentRequest

content = stack0.marketing.create_content(CreateContentRequest(
    project_slug="my-project",
    title="10 Tips for Better Email Deliverability",
    body="...",
    content_type="blog_post",
))

# Approval workflow
stack0.marketing.approve_content(ApproveContentRequest(id=content.id))

from stack0.marketing.types import CreateScriptRequest, ScheduleContentRequest

script = stack0.marketing.create_script(CreateScriptRequest(
    project_slug="my-project",
    name="Product Demo Script",
    content="...",
))

stack0.marketing.schedule_content(ScheduleContentRequest(
    content_id=content.id,
    scheduled_at="2025-03-01T09:00:00Z",
    channel="blog",
))

from stack0._exceptions import (
    Stack0Error,
    APIError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError,
    TimeoutError,
    NetworkError,
)

try:
    stack0.mail.send(request)
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except NotFoundError:
    print("Resource not found")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except APIError as e:
    print(f"API error {e.status_code}: {e.message}")
    if e.code:
        print(f"Error code: {e.code}")
except TimeoutError:
    print("Request timed out")
except NetworkError:
    print("Network connectivity issue")
except Stack0Error as e:
    print(f"SDK error: {e}")

from stack0._types import Environment, BatchJobStatus, ScheduleFrequency

# Environment
env: Environment = "sandbox"       # or "production"

# BatchJobStatus
status: BatchJobStatus = "pending"  # "pending" | "processing" | "completed" | "failed" | "cancelled"

# ScheduleFrequency
freq: ScheduleFrequency = "daily"   # "hourly" | "daily" | "weekly" | "monthly"