Time Workday RaaS Requests

uv run time_requests.py -h

usage: time_requests.py [-h] [--url URL] [--user USER] [--count COUNT] [--timeout TIMEOUT] [--download]

Time a Workday RaaS request.

options:
  -h, --help         show this help message and exit
  --url URL          The target URL to request.
  --user USER        Workday ISU Username
  --count COUNT      Number of times to run the request (default=1)
  --timeout TIMEOUT  Request timeout in minutes (default=1)
  --download         Save the output as a timestamped CSV file

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "requests",
# ]
# ///
#!/usr/bin/env python

from __future__ import annotations

import argparse
import datetime
import getpass
import os
import time
from collections.abc import Sequence
from pathlib import Path

import requests
from requests.auth import HTTPBasicAuth

# Default configuration
DEFAULT_URL = "TODO"


def log(message: str) -> None:
    """Helper to print messages with a timestamp."""
    print(f"[{datetime.datetime.now().isoformat()}] {message}")


def format_size(bytes_size: int) -> str:
    """Formats bytes into a human-readable string with commas."""
    mb_size = bytes_size / (1024 * 1024)
    return f"{bytes_size:,} bytes ({mb_size:,.2f} MB)"


def time_http_request(
    url: str,
    username: str,
    password: str,
    timeout_minutes: int = 1,
    download: bool = False,
) -> datetime.timedelta | None:
    """Performs the HTTP request, measures duration, and optionally saves the file."""

    start_time = datetime.datetime.now()
    log("Timer started. Sending request...")

    try:
        response = requests.get(
            url,
            auth=HTTPBasicAuth(username, password),
            timeout=timeout_minutes * 60,
            stream=True,
        )
        response.raise_for_status()

        line_count = 0
        total_bytes = 0

        if download:
            # Generate filename based on current timestamp
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"{timestamp}.csv"
            filepath = Path(__file__).parent / filename

            # Download the file
            log(f"Downloading content to: {filename}")
            with open(filepath, "wb") as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
            log("Download complete.")

            # Get file size and line count from the saved file
            total_bytes = os.path.getsize(filepath)
            with open(filepath, "rb") as f:
                while chunk := f.read(1024 * 1024):
                    line_count += chunk.count(b"\n")
        else:
            # Consume content to ensure request is fully complete without saving
            # We use iter_lines to count lines, but track raw length for size
            for line in response.iter_lines(decode_unicode=True):
                if line:
                    line_count += 1
                    # Approximate size for in-memory streaming
                    total_bytes += len(line.encode("utf-8")) + 1

        log(f"Line Count: {line_count:,}")
        log(f"File Size:  {format_size(total_bytes)}")

        end_time = datetime.datetime.now()
        duration = end_time - start_time

        log(f"Request completed. Status Code: {response.status_code}")
        log(f"Total Duration: {duration}")
        return duration

    except Exception as e:
        fail_time = datetime.datetime.now()
        duration = fail_time - start_time
        log(f"Request FAILED. Error: {e}")
        log(f"Total Duration until failure: {duration}")

        sleep_duration = (timeout_minutes / 2) * 60
        log(f"Sleeping for {timeout_minutes / 2} minutes before continuing...")
        time.sleep(sleep_duration)
        return duration


def main(argv: Sequence[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Time a Workday RaaS request.")
    parser.add_argument("--url", default=DEFAULT_URL, help="The target URL to request.")
    parser.add_argument("--user", help="Workday ISU Username")
    parser.add_argument(
        "--count",
        type=int,
        default=1,
        help="Number of times to run the request (default=1)",
    )
    parser.add_argument(
        "--timeout", type=int, default=1, help="Request timeout in minutes (default=1)"
    )
    parser.add_argument(
        "--download",
        action="store_true",
        help="Save the output as a timestamped CSV file",
    )

    args = parser.parse_args(argv)

    # Handle Credentials
    username = args.user or input("Enter Username: ")
    password = getpass.getpass("Enter Password: ")
    url = args.url or input("Enter URL: ")

    results: list[datetime.timedelta | None] = []

    # Request/download the report n times
    for i in range(1, args.count + 1):
        print(f"\nPreparing request {i:02d}/{args.count:02d}")
        print("-" * 30)

        duration = time_http_request(
            url=url,
            username=username,
            password=password,
            timeout_minutes=args.timeout,
            download=args.download if i == 1 else False,  # Download the first time only
        )
        results.append(duration)
        print("\n")

    # Final Summary in Minutes
    print("=" * 30)
    print("SUMMARY (Minutes)")
    print("=" * 30)

    formatted_results = []
    for i, delta in enumerate(results, 1):
        if delta:
            # Convert total seconds to minutes
            mins = delta.total_seconds() / 60
            formatted_results.append(f"{mins:.2f}")
            print(f"Run {i:02d}: {mins:.2f} min")
        else:
            formatted_results.append("FAILED")
            print(f"Run {i:02d}: FAILED")

    print(f"\nAll Durations (min): {formatted_results}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        print("\nProcess interrupted by user.")
        SystemExit(1)