soundsource

--- soundsource-0.1.0/src/soundsource/config.py (excerpt) ---
"""Configuration and authentication for the SDK."""

import os
from dataclasses import dataclass

DEFAULT_BASE_URL = "https://api.soundsource.example.com/v1"
DEFAULT_TIMEOUT = 30.0
DEFAULT_MAX_RETRIES = 3


@dataclass(frozen=True)
class Config:
    """SDK configuration."""

    api_key: str
    base_url: str = DEFAULT_BASE_URL
    timeout: float = DEFAULT_TIMEOUT
    max_retries: int = DEFAULT_MAX_RETRIES

    def __post_init__(self) -> None:
        if not self.api_key or not isinstance(self.api_key, str):
            raise ValueError("API key must be a non-empty string")

    @classmethod
    def from_env(cls, api_key_var: str = "SOUNDSOURCE_API_KEY") -> "Config":
        """Load configuration from environment variables."""
        api_key = os.environ.get(api_key_var)
        if not api_key:
            raise ValueError(
                f"Missing API key. Set the {api_key_var} environment variable."
            )

        return cls(
            api_key=api_key,
            base_url=os.environ.get("SOUNDSOURCE_BASE_URL", DEFAULT_BASE_URL),
            timeout=float(os.environ.get("SOUNDSOURCE_TIMEOUT", DEFAULT_TIMEOUT)),
            max_retries=int(os.environ.get("SOUNDSOURCE_MAX_RETRIES", DEFAULT_MAX_RETRIES)),
        )

    def get_headers(self) -> dict[str, str]:
        """Build request headers."""
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        

--- soundsource-0.1.0/examples/async_example.py (excerpt) ---
"""Async usage example for high-throughput applications."""

import asyncio
import os

from soundsource import AsyncSoundSource
from soundsource.config import Config
from soundsource.models import CreateJobRequest


async def process_multiple(urls: list[str]) -> None:
    config = Config(
        api_key=os.environ.get("SOUNDSOURCE_API_KEY", "your-api-key-here"),
        base_url="https://api.soundsource.app.v1",
    )

    async with AsyncSoundSource(config) as client:
        # Submit all jobs concurrently
        tasks = [
            client.create_job(CreateJobRequest(source_url=url))
            for url in urls
        ]
        jobs = await asyncio.gather(*tasks)
        print(f"Submitted {len(jobs)} jobs")

        # Wait for all to complete concurrently
        completed = await asyncio.gather(
            *[client.wait_for_completion(job.id) for job in jobs]
        )

        for job in completed:
            if job.transcription:
                print(f"{job.id}: {job.transcription.full_text[:100]}...")


async def stream_and_poll(job_id: str) -> None:
    """Demonstrate polling with async."""
    config = Config.from_env()
    async with AsyncSoundSource(config) as client:
        while True:
            job = await client.get_job(job_id)
            print(f"Status: {job.status}")
            if job.is_done:
                break
            await asyncio.sleep(2)


if __name__ == "__main__":
    urls = [
        "h

--- soundsource-0.1.0/examples/basic_usage.py (excerpt) ---
"""Basic synchronous usage example."""

import os

from soundsource import SoundSource
from soundsource.config import Config
from soundsource.models import CreateJobRequest


def main() -> None:
    # Option 1: Load from environment (SOUNDSOURCE_API_KEY)
    # client = SoundSource()

    # Option 2: Pass config explicitly
    config = Config(
        api_key=os.environ.get("SOUNDSOURCE_API_KEY", "your-api-key-here"),
        base_url="https://api.soundsource.app.v1",
    )

    with SoundSource(config) as client:
        # Check API health
        health = client.health_check()
        print(f"API Status: {health}")

        # Submit a job from a URL
        request = CreateJobRequest(
            source_url="https://soundsource.app.v1/podcast-episode.mp3",
            language_hint="en",
            metadata={"project": "demo", "user": "alice"},
        )
        job = client.create_job(request)
        print(f"Created job: {job.id} (status: {job.status})")

        # Poll until completion
        completed = client.wait_for_completion(job.id, poll_interval=1.0)
        print(f"Job finished with status: {completed.status}")

        if completed.transcription:
            print(f"Transcription: {completed.transcription.full_text}")
            print(f"Word count: {completed.transcription.word_count}")
            for seg in completed.transcription.segments:
                print(f"  [{seg.start:.2f}s - {seg.end:.2f}s] {seg.text}")