Async Fetch API

import asyncio
from tif1.async_fetch import fetch_json_async

async def fetch_data():
    # Fetch laps data for 2025 Monaco Qualifying
    data = await fetch_json_async(
        year=2025,
        gp="Monaco Grand Prix",
        session="Qualifying",
        path="laps.json"
    )
    print(f"Loaded {len(data.get('laps', []))} laps")

asyncio.run(fetch_data())

import asyncio
from tif1.async_fetch import fetch_json_async

async def fetch_with_custom_settings():
    # Fetch with aggressive timeout and no retries (ultra-fast)
    data = await fetch_json_async(
        year=2025,
        gp="Monaco Grand Prix",
        session="Race",
        path="drivers.json",
        timeout=10,
        max_retries=0
    )
    return data

asyncio.run(fetch_with_custom_settings())

import asyncio
from tif1.async_fetch import fetch_json_async

async def fetch_fresh_data():
    # Force fresh data from CDN (skip cache)
    data = await fetch_json_async(
        year=2025,
        gp="Monaco Grand Prix",
        session="Race",
        path="weather.json",
        use_cache=False,
        write_cache=True  # Still cache the fresh data
    )
    return data

asyncio.run(fetch_fresh_data())

import asyncio
from tif1.async_fetch import fetch_json_async

async def fetch_without_validation():
    # Skip validation for ~10-20% faster parsing
    data = await fetch_json_async(
        year=2025,
        gp="Monaco Grand Prix",
        session="Race",
        path="laps.json",
        validate_payload=False
    )
    return data

asyncio.run(fetch_without_validation())

import asyncio
from tif1.async_fetch import fetch_json_async
from tif1.exceptions import NetworkError, DataNotFoundError, InvalidDataError

async def fetch_with_error_handling():
    try:
        data = await fetch_json_async(
            year=2025,
            gp="Monaco Grand Prix",
            session="Race",
            path="laps.json"
        )
        return data
    except DataNotFoundError as e:
        # Resource doesn't exist (404)
        print(f"Data not found: {e}")
        print(f"Year: {e.context.get('year')}")
        print(f"Event: {e.context.get('event')}")
        print(f"Session: {e.context.get('session')}")
    except InvalidDataError as e:
        # JSON parsing or validation failed
        print(f"Invalid data: {e}")
        print(f"Reason: {e.context.get('reason')}")
    except NetworkError as e:
        # All retry attempts failed
        print(f"Network error: {e}")
        print(f"URL: {e.context.get('url')}")
        print(f"Status: {e.context.get('status_code')}")

asyncio.run(fetch_with_error_handling())

import asyncio
from tif1.async_fetch import fetch_multiple_async

async def fetch_driver_laps():
    # Fetch laps for multiple drivers in parallel
    drivers = ["VER", "HAM", "LEC", "SAI", "NOR", "PIA", "RUS", "ALO"]
    requests = [
        (2025, "Monaco Grand Prix", "Qualifying", f"laps/{driver}.json")
        for driver in drivers
    ]

    results = await fetch_multiple_async(requests)

    # Process results
    for driver, data in zip(drivers, results):
        if data is not None:
            laps = data.get("laps", [])
            print(f"{driver}: {len(laps)} laps")
        else:
            print(f"{driver}: No data available")

asyncio.run(fetch_driver_laps())

import asyncio
from tif1.async_fetch import fetch_multiple_async

async def fetch_driver_telemetry():
    # Fetch telemetry for all laps of a driver
    driver = "VER"
    num_laps = 50

    requests = [
        (2025, "Monaco Grand Prix", "Race", f"telemetry/{driver}_{lap}_tel.json")
        for lap in range(1, num_laps + 1)
    ]

    # Use higher concurrency for telemetry (larger batch)
    results = await fetch_multiple_async(
        requests,
        max_concurrent_requests=50
    )

    # Count successful fetches
    successful = sum(1 for r in results if r is not None)
    print(f"Fetched {successful}/{num_laps} telemetry files")

asyncio.run(fetch_driver_telemetry())

import asyncio
from tif1.async_fetch import fetch_multiple_async

async def fetch_weekend_data():
    # Fetch laps for all sessions in a race weekend
    sessions = ["Practice 1", "Practice 2", "Practice 3", "Qualifying", "Race"]
    requests = [
        (2025, "Monaco Grand Prix", session, "laps.json")
        for session in sessions
    ]

    results = await fetch_multiple_async(requests)

    # Process results
    for session, data in zip(sessions, results):
        if data is not None:
            laps = data.get("laps", [])
            print(f"{session}: {len(laps)} laps")
        else:
            print(f"{session}: Not available")

asyncio.run(fetch_weekend_data())

import asyncio
from tif1.async_fetch import fetch_multiple_async

async def fetch_with_custom_settings():
    # Fetch with custom concurrency and timeout
    requests = [
        (2025, "Monaco Grand Prix", "Race", f"laps/driver{i}.json")
        for i in range(100)
    ]

    results = await fetch_multiple_async(
        requests,
        max_concurrent_requests=50,  # Higher concurrency
        timeout=15,  # Shorter timeout
        max_retries=1  # Fewer retries for speed
    )

    successful = sum(1 for r in results if r is not None)
    print(f"Success rate: {successful}/{len(requests)}")

asyncio.run(fetch_with_custom_settings())

import asyncio
from tif1.async_fetch import fetch_multiple_async

async def fetch_fresh_data():
    # Force fresh data from CDN for all requests
    requests = [
        (2025, "Monaco Grand Prix", "Race", "weather.json"),
        (2025, "Monaco Grand Prix", "Race", "rcm.json"),
        (2025, "Monaco Grand Prix", "Race", "drivers.json"),
    ]

    results = await fetch_multiple_async(
        requests,
        use_cache=False,  # Skip cache reads
        write_cache=True  # Still cache the fresh data
    )

    return results

asyncio.run(fetch_fresh_data())

import asyncio
from tif1.async_fetch import fetch_with_rate_limit, fetch_json_async

async def controlled_fetch():
    # Limit to 5 concurrent requests
    semaphore = asyncio.Semaphore(5)

    result = await fetch_with_rate_limit(
        fetch_json_async,
        2025, "Monaco Grand Prix", "Race", "drivers.json",
        semaphore=semaphore
    )
    return result

asyncio.run(controlled_fetch())

import asyncio
from tif1.async_fetch import fetch_with_rate_limit, fetch_json_async

async def fetch_with_shared_limit():
    # Share a semaphore across multiple operations
    semaphore = asyncio.Semaphore(10)

    # All these operations share the same concurrency limit
    tasks = [
        fetch_with_rate_limit(
            fetch_json_async,
            2025, "Monaco Grand Prix", "Race", f"laps/driver{i}.json",
            semaphore=semaphore
        )
        for i in range(50)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

asyncio.run(fetch_with_shared_limit())

import asyncio
from tif1.async_fetch import fetch_with_rate_limit

async def custom_operation(value: int) -> int:
    await asyncio.sleep(0.1)  # Simulate work
    return value * 2

async def rate_limited_operations():
    # Rate limit custom async operations
    semaphore = asyncio.Semaphore(5)

    tasks = [
        fetch_with_rate_limit(custom_operation, i, semaphore=semaphore)
        for i in range(20)
    ]

    results = await asyncio.gather(*tasks)
    return results

asyncio.run(rate_limited_operations())

import tif1
from tif1.async_fetch import cleanup_resources
import atexit

# Register cleanup on exit
atexit.register(cleanup_resources)

# Your application code
session = tif1.get_session(2025, "Monaco Grand Prix", "Race")
session.load(laps=True, telemetry=True)

# Cleanup happens automatically on exit
# Or call explicitly:
cleanup_resources()

import pytest
from tif1.async_fetch import cleanup_resources

@pytest.fixture(scope="function")
def tif1_session():
    import tif1
    session = tif1.get_session(2025, "Monaco Grand Prix", "Race")
    yield session
    # Clean up after each test
    cleanup_resources()

def test_fetch_data(tif1_session):
    tif1_session.load(laps=True)
    assert len(tif1_session.laps) > 0

from contextlib import contextmanager
from tif1.async_fetch import cleanup_resources
import tif1

@contextmanager
def tif1_context():
    try:
        yield tif1
    finally:
        cleanup_resources()

# Use with context manager
with tif1_context() as tif1:
    session = tif1.get_session(2025, "Monaco Grand Prix", "Race")
    session.load(laps=True)
# Cleanup happens automatically