Async Programming - Home Assistant

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

Performance: No task creation overhead
Immediate execution: Runs right away, not scheduled
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

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

Use eager tasks - They complete faster when possible
Batch operations - Reduce event loop overhead
Cache expensive results - Avoid redundant async operations
Use executor for blocking code - Keep event loop responsive
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.

​The Event Loop

​Thread Safety

​The @callback Decorator

​When to Use @callback

​Benefits of @callback

​Creating Tasks

​Basic Task Creation

​Eager Task Execution

​Task Types

​Regular Tasks

​Background Tasks

​HassJob: Job Type Detection

​Job Type Optimization

​Executor Jobs

​When to Use Executor

​Async Context Managers

​Timeout Management

​Thread-Safe Operations

​run_callback_threadsafe

​Async Patterns

​Concurrent Operations

​Limited Concurrency

​Async Initialization

​Async Utilities

​Semaphores for Rate Limiting

​Locks for Resource Protection

​Shutdown Handling

​Graceful Shutdown

​Cancel on Shutdown

​Best Practices

​1. Always Use Async APIs

​2. Use @callback for Synchronous Code

​3. Create Tasks for Async Work in Callbacks

​4. Handle Exceptions

​5. Use Appropriate Task Types

​Performance Tips

​Common Pitfalls

​Debugging Async Issues

​Enable Async Debugging

​Find Blocking Code

​Related Resources

The Event Loop

Thread Safety

The @callback Decorator

When to Use @callback

Benefits of @callback

Creating Tasks

Basic Task Creation

Eager Task Execution

Task Types

Regular Tasks

Background Tasks

HassJob: Job Type Detection

Job Type Optimization

Executor Jobs

When to Use Executor

Async Context Managers

Timeout Management

Thread-Safe Operations

run_callback_threadsafe

Async Patterns

Concurrent Operations

Limited Concurrency

Async Initialization

Async Utilities

Semaphores for Rate Limiting

Locks for Resource Protection

Shutdown Handling

Graceful Shutdown

Cancel on Shutdown

Best Practices

1. Always Use Async APIs

2. Use @callback for Synchronous Code

3. Create Tasks for Async Work in Callbacks

4. Handle Exceptions

5. Use Appropriate Task Types

Performance Tips

Common Pitfalls

Debugging Async Issues

Enable Async Debugging

Find Blocking Code

Related Resources