Skip to content

README.md

Latest commit

 

History

History
253 lines (186 loc) · 5.91 KB

README.md

File metadata and controls

253 lines (186 loc) · 5.91 KB

CycleTask Python SDK

The official Python SDK for the CycleTask Mail Forwarding API.

PyPI version Python License: MIT

Installation

pip install cycletask

Quick Start

from cycletask import CycleTask

client = CycleTask(api_key="ct_live_your_api_key")

# List your domains
domains = client.domains.list()
print(domains)

# Create a forwarding alias
alias = client.aliases.create(
    alias="hello",
    domain_id="your_domain_id",
    destinations=["me@gmail.com"],
)
print(alias)

Async Support

import asyncio
from cycletask import AsyncCycleTask

async def main():
    client = AsyncCycleTask(api_key="ct_live_your_api_key")

    domains = await client.domains.list()
    print(domains)

    await client.close()

asyncio.run(main())

Configuration

client = CycleTask(
    api_key="ct_live_your_api_key",
    base_url="https://api.task-cycle.com/api",  # default
    timeout=30.0,                                # seconds, default
    max_retries=3,                               # default
)
Parameter Type Default Description
api_key str None Your API key. Required for authenticated endpoints.
base_url str https://api.task-cycle.com/api API base URL.
timeout float 30.0 Request timeout in seconds.
max_retries int 3 Max retries on 429/5xx with exponential backoff.

API Reference

Domains

Manage your email domains.

# List all domains
domains = client.domains.list()

# Create a domain
domain = client.domains.create(domain="example.com")

# Get domain details
domain = client.domains.get("domain_id")

# Delete a domain
client.domains.delete("domain_id")

# Trigger DNS verification
result = client.domains.verify("domain_id")

# Get required DNS records
records = client.domains.get_dns_records("domain_id")

Aliases

Manage email forwarding aliases.

# List aliases (with pagination and filtering)
aliases = client.aliases.list(
    page=1,
    limit=10,
    search="hello",
    domain="domain_id",
    status="active",
    sort="created_at:desc",
)

# Create an alias
alias = client.aliases.create(
    alias="hello",
    domain_id="domain_id",
    destinations=["me@gmail.com", "backup@gmail.com"],
    description="My forwarding alias",
    privacy_mode=False,
)

# Update an alias
client.aliases.update("alias_id", is_active=False)
client.aliases.update("alias_id", destinations=["new@gmail.com"])

# Delete an alias
client.aliases.delete("alias_id")

# Get forwarding logs for an alias
logs = client.aliases.get_logs("alias_id", page=1, limit=50)

Stats

Access dashboard analytics and email logs.

# Dashboard overview
overview = client.stats.overview()

# Chart data (email volume over time)
chart = client.stats.chart(days=30)

# Global email logs
logs = client.stats.logs(
    page=1,
    limit=20,
    status="delivered",
    search="sender@example.com",
)

Status (Public)

Check service health. These endpoints do not require authentication.

# Can create a client without an API key for status checks
client = CycleTask()

# Service health
status = client.status.get()

# 90-day uptime history
uptime = client.status.uptime()

# Incident reports
incidents = client.status.incidents()

Error Handling

The SDK raises typed exceptions for different error conditions:

from cycletask import (
    CycleTask,
    CycleTaskError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

client = CycleTask(api_key="ct_live_your_api_key")

try:
    domain = client.domains.get("nonexistent_id")
except AuthenticationError as e:
    print(f"Bad API key: {e.message}")
except NotFoundError as e:
    print(f"Not found: {e.message}")
except ValidationError as e:
    print(f"Invalid input: {e.message}")
    print(f"Field errors: {e.errors}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except CycleTaskError as e:
    print(f"API error {e.status_code}: {e.message}")

Exception Hierarchy

Exception HTTP Status Description
CycleTaskError — Base exception for all errors
AuthenticationError 401 Invalid or missing API key
PermissionError 403 Insufficient permissions
NotFoundError 404 Resource not found
ValidationError 400, 422 Request validation failed
ConflictError 409 Resource conflict
RateLimitError 429 Rate limit exceeded
InternalServerError 5xx Server-side error
ConnectionError — Network connection failure
TimeoutError — Request timed out

Context Manager

Both sync and async clients support context managers for automatic cleanup:

# Sync
with CycleTask(api_key="ct_live_your_api_key") as client:
    domains = client.domains.list()

# Async
async with AsyncCycleTask(api_key="ct_live_your_api_key") as client:
    domains = await client.domains.list()

Automatic Retries

The SDK automatically retries requests that fail with:

  • 429 (Rate Limited) — respects the Retry-After header
  • 500, 502, 503, 504 (Server Errors)

Retries use exponential backoff starting at 0.5s, doubling each attempt, capped at 30s. The default maximum is 3 retry attempts.

Type Hints

The SDK ships with complete type annotations and a py.typed marker for PEP 561 compliance. All response types are available as TypedDict definitions:

from cycletask import Domain, Alias, StatsOverview

Requirements

  • Python 3.8+
  • httpx >= 0.24.0

License

MIT — see LICENSE for details.