Using our SDK

import time
from typing import List, Tuple

from parallel_beta import Parallel
from rich.console import Console
from rich.live import Live
from rich.table import Table

PARALLEL_API_KEY = "your-api-key"
task_spec = """
Find the most recent official budget document for the specified city/town.

City Name: ${City_Name}
State: ${State}
Domain (if provided): ${Domain}

Requirements:

1. The budget document must be from an official government source
2. Verify the document is the most recent available budget
3. Ensure the URL points directly to the budget document (PDF or similar)
4. If multiple budget documents exist (e.g. proposed vs adopted), prefer the final/adopted version

Please:

1. Search the official city/town government website
2. Verify the website is authentic (proper .gov domain or matches provided domain)
3. Navigate to the finance/budget section
4. Locate the most recent budget document
5. Verify the document is an official budget (not a draft/proposal unless that's the only available version)
"""
console = Console()

def create_city_budget_task(client: Parallel):
    """Creates a task for finding city budget documents using the Parallel SDK"""

    task = client.tasks.create(
        name="City Budget Document Finder",
        description="Locate official municipal budget documents from government sources",
        processor="neon",
        prompt=task_spec,
        input_schema={
            "properties": {
                "City_Name": {"type": "string", "description": "Name of the city"},
                "State": {
                    "type": "string",
                    "description": "State where the city is located",
                },
                "Domain": {
                    "type": "string",
                    "description": "Official domain of the city government",
                },
            },
            "required": ["City_Name", "State"],
            "type": "object",
        },


        output_schema={
            "properties": {
                "budget_year": {
                    "type": "string",
                    "description": "The year of the most recent budget document",
                },
                "budget_url": {
                    "type": "string",
                    "description": "Direct URL to the official budget document",
                },
                "source_validity": {
                    "type": "string",
                    "description": "Explanation of why this is an official source",
                },
                "last_updated": {
                    "type": "string",
                    "description": "When this information was last verified (YYYY/MM)",
                },
            },
            "type": "object",
        },
        timeout=10,
    )
    return task

def run_budget_search(client: Parallel, cities: List[Tuple[str, str, str]]):
    """Run budget document search for multiple cities"""

    # Create the task
    task = create_city_budget_task(client)
    results = []

    # Run task for each city
    for city_name, state, domain in cities:
        try:
            # Enqueue task runs to execute in parallel
            run = client.tasks.runs.create(
                task_id=task.task_id,
                arguments={"City_Name": city_name, "State": state, "Domain": domain},
                runner="neon",
                enqueue_run=True,
                timeout=10,
            )
            results.append(run)
        except Exception as e:
            console.print(f"[red]Error processing {city_name}, {state}: {str(e)}[/]")

    return results, task.task_id

def display_results(results: List[dict], cities: List[Tuple[str, str, str]]):
    """Display results in a formatted table"""

    table = Table(show_header=True, header_style="bold magenta")
    table.add_column("City")
    table.add_column("State")
    table.add_column("Budget Year")
    table.add_column("URL")
    table.add_column("Last Updated")
    table.add_column("Status")
    table.add_column("Run ID")

    for idx, result in enumerate(results):
        city_name, state, _ = cities[idx]
        try:
            default_value = "N/A" if "complete" in result.status else "Pending"
            output = result.output or {}
            table.add_row(
                city_name,
                state,
                output.get("budget_year", default_value),
                output.get("budget_url", default_value),
                output.get("last_updated", default_value),
                result.status,
                result.run_id,
            )
        except Exception as e:
            table.add_row(
                city_name, state, "Error", str(e), default_value, "failed", "N/A"
            )
    return table

def main():
    # Initialize client
    client = Parallel(api_key=PARALLEL_API_KEY)

    # Define test cities
    test_cities = [
        ("City of Fairhope", "AL", "fairhopeal.gov"),
        ("Charter Township of Fenton", "MI", "fentontownship.org"),
        ("City of Foley", "AL", "cityoffoley.org"),
        # ... add more cities as needed
    ]

    console.print("[bold green]Starting budget document search...[/]")

    # Run the search
    results, task_id = run_budget_search(client, test_cities)
    run_ids = [result.run_id for result in results]

    console.print(f"\n[bold blue]Results for task id {task_id}:[/]")
    # Poll for results and update a table in real time
    with Live(display_results(results, test_cities), refresh_per_second=1) as live:
        while not all("complete" in result.status for result in results):
            time.sleep(1)
            results = [
                client.tasks.runs.retrieve(task_id=task_id, run_id=run_id)
                for run_id in run_ids
            ]
            live.update(display_results(results, test_cities))
        # keep the live view open so the complete table can auto-resize
        console.input()

if **name** == "**main**":
    main()