Async Python Basics: How to Write Scripts That Reduce Wait Times

Suresh Vina

Mar 30, 2026

What is async Python?

Before we get into code or specific examples, it helps to understand what async, or asynchronous, Python does and why it exists.

Normally when you write a Python script, everything runs line by line in the order it's called. Each task waits for the previous one to finish before moving on. This is synchronous execution.

For most scripts this works fine, but imagine you need to make 50 API calls to an external service. With sync code, each call has to be completed before the next one starts. Your script spends most of its time just waiting for responses.

Async Python attempts to solve this waiting around. With async, your script can start a task, and while it's waiting for a response, move on to starting the next task. When the first response comes back, it picks up where it left off.

Async is actually quite similar to how threads work in Python, but instead of the operating system deciding when to switch between tasks, you control that explicitly with await. This makes async particularly well-suited for tasks that involve a lot of network I/O, where your code spends most of its time waiting for responses.

Python the synchronous way

We'll start with a simple example using sync Python where we run two tasks: boiling the kettle and making toast. With sync Python, we can only do one thing at a time, so we boil the kettle first, and only once it's done do we move on to making the toast.

To simulate waiting, we're using time.sleep(). Think of this as standing in for any real-world waiting, such as waiting for an API call to return or a file to finish loading.

import time

def boil_kettle():
    print("Kettle on...")
    time.sleep(3)
    print("Kettle boiled")

def make_toast():
    print("Bread in toaster...")
    time.sleep(2)
    print("Toast ready")

def main():
    start = time.time()
    boil_kettle()
    make_toast()
    end = time.time()
    print(f"Total time: {end - start:.2f} seconds")

if __name__ == "__main__":
    main()

# output

Kettle on...
Kettle boiled
Bread in toaster...
Toast ready
Total time: 5.01 seconds

This is how we'd typically write the script in Python. Each task runs one after the other. The total time here is 5 seconds because time.sleep() blocks the entire program. While the kettle is boiling, nothing else can run. The script just sits there waiting, even though it could have started the toast in the meantime.

But the kettle and the toast don't depend on each other. There's no reason to wait for the kettle before starting the toast. In this example, 5 seconds isn't a big deal. But imagine you have 20 tasks like this. The waiting time adds up quickly, and your script ends up spending most of its time doing nothing, just waiting for the previous task to finish before it can start the next one.

Python the asynchronous way

Now let's rewrite the same example using async Python. The goal is to kick off both tasks without waiting for one to finish before starting the other, and let each one complete when it's ready.

import asyncio
import time

async def boil_kettle():
    print("Kettle on...")
    await asyncio.sleep(3)
    print("Kettle boiled!")

async def make_toast():
    print("Bread in toaster...")
    await asyncio.sleep(2)
    print("Toast ready!")

async def make_breakfast():
    start = time.time()
    await asyncio.gather(boil_kettle(), make_toast())
    end = time.time()
    print(f"Total time: {end - start:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(make_breakfast())

#output

Kettle on...
Bread in toaster...
Toast ready!
Kettle boiled!
Total time: 3.00 seconds

The total time is now 3 seconds instead of 5. Both tasks kicked off without waiting for the other, and each one finished when it was ready. Don't worry too much about the syntax just yet. Before we break down the code, let's go over the key async concepts that will make everything click once we come back to the example.

Key async Python concepts

Let's go over the key concepts and syntax you'll come across when working with async Python.

Coroutine: A function defined with async def. Unlike a regular function that runs to completion when called, a coroutine returns a coroutine object when called. It only runs when it's awaited or scheduled by the event loop.
Event Loop: The core of any async Python program, responsible for managing and running coroutines. It keeps track of all the tasks that are running, paused, or waiting, and decides which one to run next. You don't interact with the event loop directly in most cases. asyncio.run() handles that for you.
await: Pauses the current coroutine and hands control back to the event loop. The event loop can then run other coroutines while the current one is waiting. As mentioned earlier, await can only be used inside an async def function, and only with awaitables.
asyncio.gather(): Schedules multiple coroutines to run concurrently and waits for all of them to complete. The results are returned in the same order as the coroutines were passed in.
asyncio.run(): The entry point for an async program. It creates the event loop, runs the coroutine you pass to it, and closes the loop when it's done. You call it once, usually at the bottom of your script.

Now, let's go back and look at the async example and walk through what each part is doing.

Async def defines a coroutine. A coroutine is just a special kind of function that can be paused and resumed. When a coroutine hits an await, it pauses and hands control back to the event loop, allowing other coroutines to run in the meantime.

import asyncio
import time

async def boil_kettle():
    print("Kettle on...")
    await asyncio.sleep(3)
    print("Kettle boiled!")

async def make_toast():
    print("Bread in toaster...")
    await asyncio.sleep(2)
    print("Toast ready!")

async def make_breakfast():
    start = time.time()
    await asyncio.gather(boil_kettle(), make_toast())
    end = time.time()
    print(f"Total time: {end - start:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(make_breakfast())

Await can only be used inside an async def function, and it can only be used with something that's awaitable. An awaitable is simply an object that supports being awaited. This is why we use asyncio.sleep() instead of time.sleep(). The regular time.sleep() blocks the entire program and doesn't give control back to the event loop. Asyncio.sleep() is awaitable, so when it's called, the event loop can move on and run other tasks while waiting.

Asyncio.gather() takes multiple coroutines and runs them concurrently, waiting for all of them to finish before moving on.

Asyncio.run() is the entry point for any async program. It starts the event loop and runs the coroutine you pass to it. You typically call it once at the bottom of your script.

Please note that async isn't always the right tool. It's suitable when your program spends a lot of time waiting on things outside of Python, such as network requests, API calls, or database queries. If your script is doing heavy computation, async won't help much because the bottleneck is the CPU, not the waiting.

A good rule of thumb is to use async when:

Your tasks spend most of their time waiting for a response, such as making multiple API calls, and you want to avoid sitting idle between each one.
You're building something that needs to handle many operations without blocking, such as a script that queries multiple endpoints at the same time.

Using async Python with the Infrahub SDK

Now that we have a good understanding of async Python, let's bring it into the context of Infrahub. The Infrahub Python SDK supports both sync and async out of the box. The good news is that switching from one to the other is not a big change. The methods are the same, the arguments are the same, and the logic is the same. The only differences are the client class you import, the async def and await keywords, and wrapping everything in asyncio.run().

For the examples in this post, we'll be using the Infrahub sandbox, where you can log in using the default credentials. Once logged in, navigate to Account Settings > Tokens to generate an API token. (The sandbox resets daily so if your token stops working you'll need to generate a new one.)

Let's look at creating a VLAN using both versions side by side. The sync version is straightforward. You create the client, call client.create(), then call vlan.save(). Everything runs top to bottom, and each line waits for the previous one to finish.

from infrahub_sdk import InfrahubClientSync, Config

config = Config(
    address="https://sandbox.infrahub.app",
    api_token="189c12c6-a5cf-dd11-cf8e-1065d2babe57",
)

client = InfrahubClientSync(config=config)

vlan = client.create(
    kind="InfraVLAN", name="myvlan", vlan_id=100, status="provisioning", role="user"
)
vlan.save()

The async version uses InfrahubClient instead of InfrahubClientSync. The client is created inside an async def function, and every SDK call has await in front of it. The await tells Python that this call involves waiting, so the event loop is free to do other things in the meantime. Finally, asyncio.run(main()) starts the event loop and runs everything.

from infrahub_sdk import Config, InfrahubClient
import asyncio

config = Config(
    address="https://sandbox.infrahub.app",
    api_token="189c12c6-a5cf-dd11-cf8e-1065d2babe57",
)

async def main():
    client = InfrahubClient(config=config)
    vlan = await client.create(
        kind="InfraVLAN", name="myvlan", vlan_id=100, status="provisioning", role="user"
    )
    await vlan.save()

asyncio.run(main())

For a single VLAN, you won't notice any difference in speed. But as we saw in the earlier example, if you're creating or querying multiple resources at the same time, the async version pulls ahead because it doesn't sit and wait for each operation to complete before starting the next one.

How to organize async code into multiple functions

In the previous example, we looked at the basic difference between the sync and async client and how the same operations are written in each. In this example, we'll take it a step further by breaking the logic into separate async functions. This is a more realistic pattern you would use in a real script, where each operation has its own function, and the main function simply orchestrates the calls.

from infrahub_sdk import Config, InfrahubClient
import asyncio

config = Config(
    address="https://sandbox.infrahub.app",
    api_token="189c12c6-a5cf-dd11-cf8e-1065d2babe57",
)

async def get_site(client, site_name):
    return await client.get(kind="LocationSite", name__value=site_name)

async def create_vlan(client, name, vlan_id, site):
    vlan = await client.create(
        kind="InfraVLAN", name=name, vlan_id=vlan_id, status="provisioning", role="user", site=site
    )
    await vlan.save()
    return vlan

async def main():
    client = InfrahubClient(config=config)
    site = await get_site(client, "atl1")
    vlan = await create_vlan(client, "myvlan", 100, site)
    print(f"Created VLAN {vlan.name.value} at site {site.name.value}")

asyncio.run(main())

We have two functions—get_site() and create_vlan()—each responsible for a single operation. Both accept the client as an argument so we're not creating a new client inside every function. The main() function ties everything together, first fetching the site and then passing it to create_vlan(). Since the VLAN creation depends on the site being fetched first, both calls use sequential await rather than asyncio.gather().

When tasks depend on each other, sequential await is the right approach. The site must be fetched before it can be passed to the VLAN creation, so we wait for one to finish before starting the next.

Asyncio.gather() is for tasks that are independent of each other. If you need to fetch multiple sites at the same time, or create multiple VLANs that don't depend on each other, that is where asyncio.gather() makes sense. It kicks off all the tasks at once and waits for all of them to finish, rather than waiting for each one individually.

How to run the same operation across multiple sites at once

In the previous example, we fetched a single site and created a single VLAN using sequential await. In this example, we'll look at a more realistic scenario where we need to perform the same operation across multiple sites.

from infrahub_sdk import Config, InfrahubClient
import asyncio

config = Config(
    address="https://sandbox.infrahub.app",
    api_token="189c12c6-a5cf-dd11-cf8e-1065d2babe57",
)

async def get_site(client, site_name):
    return await client.get(kind="LocationSite", name__value=site_name)

async def create_vlan(client, site):
    vlan = await client.create(
        kind="InfraVLAN", name=f"myvlan_{site.name.value}", vlan_id=100, status="provisioning", role="user", site=site
    )
    await vlan.save()

async def main():
    client = InfrahubClient(config=config)
    
    atl1, den1 = await asyncio.gather(
        get_site(client, "atl1"),
        get_site(client, "den1")
    )
    
    await asyncio.gather(
        create_vlan(client, atl1),
        create_vlan(client, den1)
    )

asyncio.run(main())

We fetch both sites concurrently using asyncio.gather(), since neither depends on the other. Once we have both sites, we use asyncio.gather() again to create the VLANs concurrently. The VLAN creation still depends on the site being fetched first, so we wait for the first gather to complete before starting the second. Within each gather, the tasks are independent of each other, so they run concurrently.

A note about awaitables and async-compatible libraries

As we briefly covered earlier, an awaitable is any object that can be used with await. In Python, there are three main types of awaitables: coroutines, tasks, and futures. When you write await client.get(), for example, the SDK method returns a coroutine, which the event loop knows how to run and pause.

This brings up an important point. Not every library works with async. A regular blocking library like requests can't be used with await. The event loop can't pause a blocking call, so while requests is waiting for a response, nothing else can run.

To make HTTP calls in an async context, you may need a library that's built with async support. The two most common options are aiohttp and httpx. Both are designed to work with await and will hand control back to the event loop while waiting for a response.

When you use the Infrahub Python SDK, you don't need to worry about any of this. The async client uses an async-compatible HTTP library under the hood, so all the SDK methods are already awaitable and work seamlessly with the event loop. But it's a point worth understanding when you start building your own async code that makes HTTP calls outside of the SDK.

Getting started with async Python

If you're ready to start writing asynchronous scripts for your own infrastructure automation, here are some next steps:

Explore the Infrahub Python SDK documentation to see the full range of async capabilities you can work with.
Try it yourself by experimenting with async queries and mutations in the Infrahub sandbox.