Async Programming

Advanced async patterns and best practices in Home Assistant Core

Home Assistant is built on Python’s asyncio framework, making it event-driven and highly concurrent. Understanding async programming patterns is essential for writing efficient integrations.

The Event Loop

Home Assistant runs a single asyncio event loop that coordinates all async operations. All integration code must respect this event loop.

Thread Safety

The event loop runs in a specific thread. Most Home Assistant operations must run in this thread:

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")

Reference: homeassistant/core.py:422

The @callback Decorator

The @callback decorator marks functions as safe to call from within the event loop:

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

Reference: homeassistant/core.py:210

When to Use @callback

• Event handlers that only do synchronous work
• State change callbacks
• Timer callbacks
• Any function called directly from the event loop

Benefits of @callback

1. Performance: No task creation overhead
2. Immediate execution: Runs right away, not scheduled
3. Guaranteed order: Executes before other scheduled tasks

Creating Tasks

Basic Task Creation

Use async_create_task to schedule coroutines:

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"
    )

Reference: homeassistant/core.py:755

Eager Task Execution

By default, Home Assistant uses eager task execution for better performance:

# 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
)

Eager tasks can complete synchronously if they don’t await, reducing overhead.

Reference: homeassistant/util/async_.py:25

Task Types

Regular Tasks

Regular tasks block startup and shutdown:

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

• Home Assistant waits for these during startup
• Prevents shutdown until complete
• Use for critical initialization

Background Tasks

Background tasks don’t block startup or shutdown:

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

• Don’t block startup
• Automatically cancelled on shutdown
• Use for long-running monitoring
• Not waited for in async_block_till_done

Reference: homeassistant/core.py:806

HassJob: Job Type Detection

Home Assistant uses HassJob to optimize job execution based on callable type:

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

Reference: homeassistant/core.py:296

Job Type Optimization

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

Reference: homeassistant/core.py:347

Executor Jobs

For CPU-intensive or blocking I/O operations, use the executor:

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"]
    )

Reference: homeassistant/core.py:838

When to Use Executor

• File I/O operations
• CPU-intensive computations
• Third-party libraries that block
• Any operation that takes > 10ms

Never perform blocking operations directly in the event loop. Always use the executor for blocking code.

Async Context Managers

Timeout Management

Home Assistant provides timeout utilities:

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")

Reference: homeassistant/core.py:414

Thread-Safe Operations

run_callback_threadsafe

Call event loop code from another thread:

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()

Reference: homeassistant/util/async_.py:52

run_callback_threadsafe cannot be called after Home Assistant begins shutdown. It will raise RuntimeError to prevent deadlocks.

Async Patterns

Concurrent Operations

Run multiple operations concurrently:

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)

Limited Concurrency

Limit concurrent operations to avoid overwhelming resources:

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)]

Reference: homeassistant/util/async_.py:100

Async Initialization

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

Async Utilities

Semaphores for Rate Limiting

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)

Locks for Resource Protection

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()

Shutdown Handling

Graceful Shutdown

Register shutdown handlers for cleanup:

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

Reference: homeassistant/core.py:1020

Cancel on Shutdown

Mark jobs as cancellable during shutdown:

from homeassistant.core import HassJob

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

Reference: homeassistant/core.py:1191

Best Practices

1. Always Use Async APIs

# 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()

2. Use @callback for Synchronous Code

# 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

3. Create Tasks for Async Work in Callbacks

@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()

4. Handle Exceptions

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")

5. Use Appropriate Task Types

# 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"
)

Performance Tips

1. Use eager tasks - They complete faster when possible
2. Batch operations - Reduce event loop overhead
3. Cache expensive results - Avoid redundant async operations
4. Use executor for blocking code - Keep event loop responsive
5. Limit concurrency - Prevent resource exhaustion

Common Pitfalls

• Never call asyncio.sleep(0) excessively - it’s not free
• Don’t create tasks for every tiny operation - use @callback
• Avoid blocking the event loop - use executor for blocking code
• Don’t forget to handle task exceptions
• Clean up resources in shutdown handlers

Debugging Async Issues

Enable Async Debugging

import asyncio
import logging

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

Find Blocking Code

Home Assistant logs slow operations automatically. Look for warnings about blocking tasks.

Related Resources

• Event Handling
• State Management
• Service Registration