Async Client and Composable Flows in the Python SDK

Overview

Relevant for Developers using asyncio or writing reusable helpers

AsyncAlertOps has exactly the same surface as AlertOps: the same resources, ops and helpers. The difference is that every call returns an awaitable and .iter() returns an async iterator. The helpers work on both clients because they're written as flows: generator functions that describe the requests to make without making them. The client runs each step, synchronously or asynchronously, and sends the result back into the generator.

Key benefits:

• Same code on both clients — every built-in helper, from groups.find to the whole semantic layer, is one flow that runs on AlertOps and AsyncAlertOps.
• Your own helpers — subclass Resource, decorate a generator with @flow, and you get the same dual behaviour.
• Normal error handling — a failed step raises inside your generator, so try/except works as usual.

Using the async client

Relevant for Developers calling AlertOps from asyncio code

Create the client inside a running event loop, preferably with async with so the connection pool closes on exit.

import asyncio
from alertops import AsyncAlertOps

async def main():
    async with AsyncAlertOps() as ao:                        # same config rules as AlertOps()
        tz = await ao.account.time_zone()                    # helper flow: awaitable
        group = await ao.groups.find("NOC")                  # helper flow: awaitable
        if group:
            members = await ao.groups.members.list(group.group_id)   # plain op: awaitable
            print(tz, group.group_name, len(members))

        async for g in ao.groups.list.iter():                # every page, async iterator
            print(g.group_id, g.group_name)

        policies = await ao.escalation_policies.list.all()   # AOList of every item

asyncio.run(main())

Async client methods

Independent calls can run concurrently with asyncio.gather, because each op is an ordinary coroutine:

async with AsyncAlertOps() as ao:
    group, user = await asyncio.gather(ao.groups.get(12345), ao.users.get(101))

Retries, Retry-After handling and error mapping behave exactly as on the sync client. Keep concurrency modest so you don't trigger 429 throttling.

What a flow is

Relevant for Developers who want to understand how the helpers work

A flow is a generator decorated with @flow. It doesn't do any I/O itself (it's "sans-IO"). Instead it yields a description of the next step, and the client's runner executes that step and sends the result back:

When a step raises (for example NotFoundError), the runner throws the exception into your generator at the yield, so a try/except around the yield catches it. If you don't catch it, it propagates to the caller as usual. The flow's return value is what the caller gets.

On AlertOps, calling a flow runs it to completion and returns the value. On AsyncAlertOps, calling the same flow returns a coroutine that you await.

This is how the built-in groups.find and groups.ensure are written:

@flow
def find(self, name):
    rows = yield self.list.lazy(search=name, limit=100)
    for g in rows:
        if str(g.get("group_name", "")).lower() == name.lower():
            return g
    return None

@flow
def ensure(self, group_name, **fields):
    existing = yield self.find.lazy(group_name)          # nested flow
    if existing is not None:
        return existing, False
    created = yield self.create.lazy(group_name=group_name, **fields)
    return created, True

Write your own flow

Relevant for Developers packaging multi-step logic

1. Subclass Resource.
2. Decorate a generator method with @flow.
3. Reach other resources through self._client and yield .lazy(...) calls, never direct calls.
4. Instantiate your resource with a client: MyOps(ao).

from alertops import AlertOps, AsyncAlertOps, NotFoundError, Resource, flow

class OnCallOps(Resource):
    @flow
    def group_size(self, name):
        """Number of members in the group called name (0 if there is no such group)."""
        g = yield self._client.groups.find.lazy(name)                      # nested flow
        if g is None:
            return 0
        members = yield self._client.groups.members.list.lazy(g.group_id)  # one request
        return len(members)

    @flow
    def group_name_or_none(self, group_id):
        """Error forwarding: a 404 raised by the step lands in this try/except."""
        try:
            g = yield self._client.groups.get.lazy(group_id)
        except NotFoundError:
            return None
        return g.group_name

    @flow
    def user_names(self, search):
        """lazy_all follows pagination and hands back every item."""
        rows = yield self._client.users.list.lazy_all(search=search)
        return [u.user_name for u in rows]

    @flow
    def sizes(self, names):
        """Yielding a list runs each step in order and returns a list of results."""
        return (yield [self.group_size.lazy(n) for n in names])


ao = AlertOps()
print(OnCallOps(ao).group_size("NOC"))                  # sync: returns the value

async def main():
    async with AsyncAlertOps() as aao:
        print(await OnCallOps(aao).group_size("NOC"))   # async: same code, awaited

A Resource subclass can also declare its own endpoints with Op, which then get .lazy(), .iter() and .all() like built-in ops:

from alertops import Op, Resource, flow

class Priorities(Resource):
    list = Op("GET", "/app/priority_types")

    @flow
    def names(self):
        rows = yield self.list.lazy()
        return [r["name"] for r in rows]

Priorities(ao).names()        # ['Critical', 'High', 'Medium', 'Low', 'No Priority']

@flow raises TypeError("@flow functions must be generators") if the function has no yield.

Built-in helper flows worth knowing

Relevant for Developers looking for an existing helper before writing one

Every helper below is a flow, so it works on both clients. docs/REFERENCE.md lists all of them under each resource's "Helpers" line.

Most *_if_exists, update, modify and patch helpers are read-merge-write wrappers that exist because many AlertOps PUTs replace the whole record.

Best Practices

Relevant for Anyone writing helpers or async services

Flows

• Write a helper as a flow whenever it makes more than one request, so it serves both clients.
• Keep flows pure: compute, yield lazy calls, return a value. Do I/O, sleeps and logging of side effects outside.
• Catch specific exceptions around the yield that can fail (NotFoundError for "already gone").
• Compose: yield other_flow.lazy(...) rather than copying its steps.

Async

• Open one AsyncAlertOps per event loop with async with.
• Use asyncio.gather for independent reads and keep the fan-out small to stay under rate limits.
• Use async for … in op.iter() for large lists rather than await op.all().

Configuration Examples

Relevant for Starting templates for common patterns

Example 1 — Get or create a group and add members (sync and async)

group, created = ao.groups.ensure("sdk-example-noc")
ao.groups.members.add_users(group.group_id, [101, 102], roles=["Primary"])

async with AsyncAlertOps() as aao:
    group, created = await aao.groups.ensure("sdk-example-noc")
    await aao.groups.members.add_users(group.group_id, [101, 102], roles=["Primary"])

Example 2 — A cleanup flow that tolerates missing objects

from alertops import NotFoundError, Resource, flow

class Cleanup(Resource):
    @flow
    def remove_groups(self, group_ids):
        removed = 0
        for gid in group_ids:
            try:
                yield self._client.groups.delete.lazy(gid)
                removed += 1
            except NotFoundError:
                pass                                  # already gone
        return removed

Cleanup(ao).remove_groups([12345, 12346])

Example 3 — Concurrent reads with the async client

import asyncio
from alertops import AsyncAlertOps

async def snapshot():
    async with AsyncAlertOps() as ao:
        groups, policies, users = await asyncio.gather(
            ao.groups.list.all(), ao.escalation_policies.list.all(), ao.lookups.users())
        return len(groups), len(policies), len(users)

print(asyncio.run(snapshot()))

Troubleshooting

Relevant for Anyone debugging async code or custom flows

"coroutine was never awaited" or a flow returns a coroutine object

• On AsyncAlertOps, every op and every flow returns a coroutine. Add await.
• Inside a flow you called an op directly instead of yielding .lazy(...). Change self._client.x.y(...) to yield self._client.x.y.lazy(...).

TypeError: @flow functions must be generators

• The decorated function has no yield. Make at least one step a yield, or drop @flow and write a plain function.

RuntimeError about the event loop, or "Event loop is closed"

• The async client was created on one loop and used on another. Create it inside the coroutine that uses it, with async with.

Calling a helper returns a generator or a Call object instead of data

• The method is a plain generator, not a @flow-decorated one, so no runner executed it. Decorate it with @flow and call it through a Resource instance bound to a client (MyOps(ao).helper()).
• You called .lazy(...) outside a flow. .lazy() only describes a step; call the op or flow directly instead.

Exception from a step isn't caught

• The try must wrap the yield itself. Exceptions are thrown into the generator at the yield of the step that failed.

Related Articles

• Getting Started with the AlertOps Python SDK
• Calling the API from Python: Parameters, Pagination and Errors
• Planning On-Call Setups with the AlertOps Semantic Layer
• Testing and Verifying the Python SDK

Need help?

Email support@alertops.com or contact your AlertOps representative.