Async Usage

Guide to asynchronous programming with HFortix using async/await.

Overview

Async support is built into the FortiOS client itself — there is no separate async class and no _async method variants. Create the client with mode="async" and every endpoint method (get(), post(), put(), delete(), set()) returns a coroutine that you await. The navigation tree and method signatures are identical in sync and async mode.

from hfortix_fortios import FortiOS

# Synchronous (default)
fgt = FortiOS(host='192.168.1.99', token='your-token')

# Asynchronous
fgt = FortiOS(host='192.168.1.99', token='your-token', mode='async')

Quick Example

import asyncio
from hfortix_fortios import FortiOS

async def main():
    fgt = FortiOS(host='192.168.1.99', token='your-token', mode='async')
    try:
        # Same methods as sync mode — just await them
        addresses = await fgt.api.cmdb.firewall.address.get()
        print(f"Found {len(addresses)} addresses")

        result = await fgt.api.cmdb.firewall.address.post(
            name='server-1',
            subnet='10.0.0.1 255.255.255.255'
        )
        print(result.http_status)
    finally:
        await fgt.aclose()

asyncio.run(main())

Async Context Manager

The client supports async with, which closes the underlying connection pool automatically:

import asyncio
from hfortix_fortios import FortiOS

async def main():
    async with FortiOS(
        host='192.168.1.99',
        token='your-token',
        mode='async'
    ) as fgt:
        status = await fgt.api.monitor.system.status.get()
        print(status['hostname'])

asyncio.run(main())

Concurrent Operations

The main benefit of async mode: run multiple API calls concurrently with asyncio.gather():

import asyncio
from hfortix_fortios import FortiOS

async def main():
    async with FortiOS(
        host='192.168.1.99',
        token='your-token',
        mode='async'
    ) as fgt:
        addresses, policies, status = await asyncio.gather(
            fgt.api.cmdb.firewall.address.get(),
            fgt.api.cmdb.firewall.policy.get(),
            fgt.api.monitor.system.status.get(),
        )
        print(f"{len(addresses)} addresses, {len(policies)} policies")
        print(f"Hostname: {status['hostname']}")

asyncio.run(main())

Error Handling in Async Code

Exceptions are the same as in sync mode (hfortix_core.exceptions) — wrap awaited calls in try/except as usual:

import asyncio
from hfortix_fortios import FortiOS
from hfortix_core.exceptions import APIError, ResourceNotFoundError

async def main():
    async with FortiOS(
        host='192.168.1.99',
        token='your-token',
        mode='async'
    ) as fgt:
        try:
            addr = await fgt.api.cmdb.firewall.address.get(name='missing')
        except ResourceNotFoundError:
            print("Address not found")
        except APIError as e:
            print(f"API error: {e}")

asyncio.run(main())

When using asyncio.gather(), pass return_exceptions=True if you want to collect failures instead of cancelling the whole batch.

Notes

Connection pooling is shared across concurrent calls; tune it with the max_connections / max_keepalive_connections constructor parameters.
Do not mix one client instance across sync and async code — pick the mode at construction time.
If you don’t use async with, call await fgt.aclose() when done.