Async Programming

from homeassistant.core import HomeAssistant

def some_integration_code(hass: HomeAssistant):
    """This will raise an error if called from wrong thread."""
    # Home Assistant verifies thread safety for critical operations
    hass.verify_event_loop_thread("my_operation")

from homeassistant.core import callback

@callback
def my_callback_function(data: dict) -> None:
    """This function is safe to call in the event loop.
    
    - No blocking I/O
    - No long-running computations
    - Executes synchronously
    """
    # Fast, synchronous operations only
    processed = data["value"] * 2
    return processed

async def my_async_operation(hass: HomeAssistant):
    """Perform async operation."""
    await asyncio.sleep(1)
    return "done"

@callback
def handle_event(event: Event):
    """Handle event and create background task."""
    hass.async_create_task(
        my_async_operation(hass),
        name="my_operation"
    )

# This starts executing immediately
task = hass.async_create_task(
    my_coroutine(),
    eager_start=True  # Default
)

# This waits until next event loop iteration
task = hass.async_create_task(
    my_coroutine(),
    eager_start=False
)

task = hass.async_create_task(
    fetch_initial_data(),
    name="fetch_data"
)

task = hass.async_create_background_task(
    monitor_device_forever(),
    name="device_monitor"
)

from homeassistant.core import HassJob, HassJobType

# Automatically detects job type
job = HassJob(my_callable, name="my_job")

# Job types:
# - Coroutinefunction: async def functions
# - Callback: @callback decorated functions  
# - Executor: Regular synchronous functions

from homeassistant.core import HassJob, HassJobType

@callback
def callback_func():
    """Fast callback."""
    pass

async def async_func():
    """Async function."""
    await asyncio.sleep(1)

def sync_func():
    """Blocking function."""
    time.sleep(1)  # BAD - blocks event loop!

# Home Assistant runs each optimally:
job1 = HassJob(callback_func)  # HassJobType.Callback - runs immediately
job2 = HassJob(async_func)     # HassJobType.Coroutinefunction - scheduled as task
job3 = HassJob(sync_func)      # HassJobType.Executor - runs in thread pool

hass.async_run_hass_job(job1)
hass.async_run_hass_job(job2)
hass.async_run_hass_job(job3)  # Runs in executor automatically

import time

def blocking_operation(data: str) -> str:
    """This blocks but runs in thread pool."""
    time.sleep(5)  # OK in executor
    return data.upper()

@callback
def handle_event(event: Event):
    """Handle event and run blocking code safely."""
    # Schedule blocking operation in executor
    hass.async_add_executor_job(
        blocking_operation,
        event.data["text"]
    )

from homeassistant.core import HomeAssistant

async def my_operation(hass: HomeAssistant):
    """Operation with timeout."""
    try:
        async with hass.timeout.async_timeout(10):
            await slow_external_api_call()
    except TimeoutError:
        _LOGGER.error("Operation timed out")

from homeassistant.util.async_ import run_callback_threadsafe
import threading

def thread_worker(hass: HomeAssistant):
    """Worker running in separate thread."""
    # Call into event loop safely
    future = run_callback_threadsafe(
        hass.loop,
        lambda: hass.states.async_set("sensor.test", "value")
    )
    # Wait for result
    future.result()

# Start worker thread
thread = threading.Thread(target=thread_worker, args=(hass,))
thread.start()

import asyncio

async def setup_platform(hass: HomeAssistant):
    """Set up platform with concurrent operations."""
    # Run multiple async operations concurrently
    results = await asyncio.gather(
        fetch_devices(),
        fetch_config(),
        fetch_user_data(),
        return_exceptions=True  # Don't fail all if one fails
    )
    
    devices, config, user_data = results
    
    # Handle any exceptions
    for result in results:
        if isinstance(result, Exception):
            _LOGGER.error("Operation failed: %s", result)

from homeassistant.util.async_ import gather_with_limited_concurrency

async def fetch_many_devices(hass: HomeAssistant):
    """Fetch many devices with concurrency limit."""
    device_ids = list(range(100))
    
    # Only 5 concurrent fetches at a time
    results = await gather_with_limited_concurrency(
        5,  # max concurrent
        *[fetch_device(device_id) for device_id in device_ids],
        return_exceptions=True
    )
    
    return [r for r in results if not isinstance(r, Exception)]

from homeassistant.core import HomeAssistant
from homeassistant.config_entries import ConfigEntry

async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry
) -> bool:
    """Set up component from config entry."""
    # Initialize async client
    client = await create_async_client(
        entry.data["host"],
        entry.data["token"]
    )
    
    # Store for later use
    hass.data[DOMAIN] = {"client": client}
    
    # Set up platforms concurrently
    await hass.config_entries.async_forward_entry_setups(
        entry,
        ["sensor", "switch", "light"]
    )
    
    return True

import asyncio

class MyIntegration:
    """Integration with rate limiting."""
    
    def __init__(self):
        # Limit to 3 concurrent API calls
        self._api_semaphore = asyncio.Semaphore(3)
    
    async def api_call(self, endpoint: str):
        """Make rate-limited API call."""
        async with self._api_semaphore:
            return await self._client.get(endpoint)

import asyncio

class DeviceManager:
    """Manage device with mutex."""
    
    def __init__(self):
        self._lock = asyncio.Lock()
        self._state = {}
    
    async def update_state(self, key: str, value: str):
        """Update state safely."""
        async with self._lock:
            # Only one update at a time
            self._state[key] = value
            await self._persist_state()

from homeassistant.core import HassJob

async def async_setup(hass: HomeAssistant):
    """Set up component with cleanup."""
    client = await create_client()
    
    async def shutdown_handler(event):
        """Clean up on shutdown."""
        await client.close()
        _LOGGER.info("Client closed")
    
    # Register shutdown handler
    hass.async_add_shutdown_job(
        HassJob(shutdown_handler)
    )
    
    return True

from homeassistant.core import HassJob

# This job will be cancelled when HA shuts down
job = HassJob(
    my_periodic_task,
    cancel_on_shutdown=True
)

# GOOD
async def fetch_data():
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

# BAD - blocks event loop
def fetch_data():
    return requests.get(url).json()

# GOOD
@callback  
def process_state_change(event: Event) -> None:
    value = event.data["new_state"].state
    return float(value) * 2

# WASTEFUL - unnecessary task creation
async def process_state_change(event: Event) -> None:
    value = event.data["new_state"].state
    return float(value) * 2

@callback
def handle_event(event: Event) -> None:
    """Handle event."""
    # Don't await in callback!
    hass.async_create_task(
        async_process_event(event)
    )

async def async_process_event(event: Event) -> None:
    """Async processing."""
    await some_async_operation()

async def robust_operation():
    """Operation with error handling."""
    try:
        await external_api_call()
    except asyncio.TimeoutError:
        _LOGGER.warning("API call timed out")
    except aiohttp.ClientError as err:
        _LOGGER.error("API error: %s", err)
    except Exception:
        _LOGGER.exception("Unexpected error")

# Startup task - blocks startup completion
hass.async_create_task(
    initialize_critical_resources(),
    name="init"
)

# Background task - doesn't block
hass.async_create_background_task(
    monitor_loop(),
    name="monitor"
)

import asyncio
import logging

# Enable asyncio debug mode
loop = asyncio.get_event_loop()
loop.set_debug(True)
logging.getLogger("asyncio").setLevel(logging.DEBUG)