Function Schedules

from datetime import datetime

from notte_sdk import NotteClient


def run(competitor_urls: list[str]):
    """Monitor competitor prices daily."""
    client = NotteClient()
    prices = []

    for url in competitor_urls:
        with client.Session() as session:
            session.execute(type="goto", url=url)
            price = session.scrape(instructions="Extract product price")
            prices.append({"url": url, "price": price})

    return {"prices": prices, "checked_at": datetime.now().isoformat()}

from datetime import datetime

from notte_sdk import NotteClient


def run(source_url: str):
    """Collect data every hour."""
    client = NotteClient()

    with client.Session() as session:
        session.execute(type="goto", url=source_url)
        data = session.scrape()

    return {"data": data, "timestamp": datetime.now().isoformat()}

from datetime import datetime

from notte_sdk import NotteClient


def run():
    """Generate weekly report."""
    client = NotteClient()

    # Collect data from multiple sources
    report_data = []

    sources = ["https://analytics.example.com", "https://dashboard.example.com"]

    for url in sources:
        with client.Session() as session:
            session.execute(type="goto", url=url)
            data = session.scrape(instructions="Extract weekly metrics")
            report_data.append(data)

    return {"report": report_data, "week": datetime.now().isocalendar()[1]}

from datetime import datetime

from notte_sdk import NotteClient


def run(target_url: str, expected_text: str):
    """Monitor site uptime."""
    client = NotteClient()

    try:
        with client.Session(idle_timeout_minutes=2) as session:
            session.execute(type="goto", url=target_url)
            content = session.scrape()

            is_up = expected_text in content

        return {"status": "up" if is_up else "down", "url": target_url, "checked_at": datetime.now().isoformat()}

    except Exception as e:
        return {"status": "error", "url": target_url, "error": str(e), "checked_at": datetime.now().isoformat()}

import os
from datetime import datetime


def run():
    # Access environment variables
    api_key = os.getenv("API_KEY")
    webhook_url = os.getenv("WEBHOOK_URL")

    # Use current date
    today = datetime.now().strftime("%Y-%m-%d")

    # Your automation
    pass

from notte_sdk import NotteClient

client = NotteClient()

# List recent runs
runs = client.functions.list_runs(
    "function_abc123",
    only_active=False,  # Include completed runs
)

for run in runs.items:
    print(f"Run {run.workflow_run_id}:")
    print(f"  Status: {run.status}")
    print(f"  Created: {run.created_at}")
    print(f"  Updated: {run.updated_at}")

def run():
    try:
        # Your automation
        result = perform_automation()

        # Send success notification
        notify_success(result)

        return result

    except Exception as e:
        # Send failure alert
        notify_failure(str(e))

        # Return error info
        return {"success": False, "error": str(e)}

def run():
    try:
        # Your automation
        result = perform_automation()
        return result

    except Exception as e:
        # Send to Slack/Discord/Email
        requests.post(
            os.getenv("WEBHOOK_URL", ""),
            json={"text": f"Scheduled function failed: {str(e)}", "function_id": "function_abc123"},
        )

        raise  # Re-raise to mark run as failed

def run():
    try:
        result = perform_automation()
        return {"success": True, "data": result}
    except Exception as e:
        # Log error, send alert
        return {"success": False, "error": str(e)}

from notte_sdk import NotteClient


def run():
    client = NotteClient()

    # Set timeout appropriate for schedule
    with client.Session(idle_timeout_minutes=10) as session:
        # Your automation
        pass

from notte_sdk import NotteClient


def run():
    """Check if all schedules are healthy."""
    client = NotteClient()

    # Get all workflow runs from last 24 hours
    workflows = client.functions.list()

    health_report = []
    for workflow in workflows.items:
        runs = client.functions.list_runs(workflow.workflow_id)

        # Check recent runs
        recent_failures = [r for r in runs.items if r.status == "failed"]

        health_report.append({"workflow_id": workflow.workflow_id, "recent_failures": len(recent_failures)})

    return health_report

def already_processed(date: str):
    """Check if date has already been processed."""
    # TODO: Implement logic to check if date has already been processed
    False


def process_data(date: str):
    """Process data for the given date."""
    # TODO: Implement logic to process data for the given date
    "processed"


def mark_processed(date: str):
    """Mark date as processed."""
    # TODO: Implement logic to mark date as processed
    "mark_as_processed"


def run(date: str):
    """Idempotent function - safe to run multiple times."""
    # Use date as unique identifier
    # Skip if already processed
    if already_processed(date):
        return {"status": "already_processed", "date": date}

    # Process data
    result = process_data(date)

    # Mark as processed
    mark_processed(date)

    return {"status": "processed", "date": date, "result": result}

from datetime import datetime

import pytz


def run():
    # Get current time in specific timezone
    tz = pytz.timezone("America/New_York")
    current_time = datetime.now(tz)

    # Your automation
    pass