TaskIQ Integration

TaskIQ Integration

Wire AsyncSteamClient / AsyncSteamPool into a TaskIQ broker via startup / shutdown hooks + a dependency function.

New in pysteam-client 1.6. TaskIQ is imported lazily — installing steam.aio doesn't force TaskIQ into your dependency graph.

Single client

from steam.aio import AsyncSteamClient
from steam.aio.integrations.taskiq import register_steam_client
from taskiq import TaskiqDepends
from taskiq_redis import ListQueueBroker

broker = ListQueueBroker(url="redis://localhost:6379")
client = AsyncSteamClient()

async def _login(c: AsyncSteamClient) -> None:
    await c.anonymous_login()

get_client = register_steam_client(broker, client, on_start=_login)

@broker.task
async def sync_app(
    app_id: int,
    steam: AsyncSteamClient = TaskiqDepends(get_client),
) -> dict:
    return await steam.get_product_info(apps=[app_id])

What register_steam_client does

1. Registers a broker startup handler that calls await client.start() then await on_start(client) if you supplied one. A partial-startup failure tears the client back down before re-raising so a failed broker startup doesn't leak a runner thread.
2. Registers a broker shutdown handler that calls await client.close().
3. Returns a sync callable () -> AsyncSteamClient that you pass to TaskiqDepends inside your @broker.task definitions. TaskIQ accepts sync deps in async tasks and this is cheaper than an async one that only reads a captured local.

Pool

from steam.aio import AsyncSteamPool, PoolMember
from steam.aio.integrations.taskiq import register_steam_pool
from taskiq import TaskiqDepends

async def _anon(c):
    await c.anonymous_login()

pool = AsyncSteamPool([
    PoolMember(account_id="a", login=_anon),
    PoolMember(account_id="b", login=_anon),
])

get_pool = register_steam_pool(broker, pool)

@broker.task
async def sync_app(
    app_id: int,
    pool: AsyncSteamPool = TaskiqDepends(get_pool),
) -> dict:
    return await pool.round_robin().get_product_info(apps=[app_id])

Pool members carry their own login callable (see PoolMember), so there's no on_start parameter on register_steam_pool.

Worker lifecycle

Each TaskIQ worker process spawns its own AsyncSteamClient instance — sockets, ACK state, and the CM session are per-process. Never share a client across processes.

The single client variable in module scope is fine because TaskIQ workers import the task module once at boot; the broker startup handler starts that worker's client, and shutdown closes it. If you're running N workers, you get N Steam CM sessions.

For fewer CM sessions than workers, terminate the CM session in each worker but route through a shared "Steam frontend" service over your own RPC — out of scope here.

Retries + typed errors

AsyncSteamClient raises typed errors — TaskIQ's retry middleware can pattern-match on them:

from taskiq import TaskiqRetryMiddleware
from steam.aio import SteamRPCTimeoutError, SteamReconnectError

broker.add_middlewares(
    TaskiqRetryMiddleware(
        default_retry_count=3,
        # Only retry on transient failures.
        no_result_on_error=False,
        retry_on=(SteamRPCTimeoutError, ConnectionError),
    ),
)

Non-transient failures (SteamLoginError, SteamNotStartedError, ValueError for bad params) propagate to the task's Exception result — those are real bugs / config issues that shouldn't retry.

Metrics through TaskIQ + prometheus

If you're using taskiq-prometheus, you already get per-task counters. Add the AsyncSteamClient metrics hook for the RPC / reconnect layer beneath the task:

from steam.aio import prometheus_hook
from prometheus_client import CollectorRegistry, generate_latest

registry = CollectorRegistry()
client = AsyncSteamClient(metrics_hook=prometheus_hook(registry=registry))

# ... register as usual ...

# Expose the registry via whatever HTTP endpoint your worker
# already runs for the standard TaskIQ metrics scraping.

Related pages

• AsyncSteamClient — the underlying client
• Pool — multi-account variant
• FastAPI-Integration — same story for FastAPI