genja-py

Python bindings for the Genja runtime.

This package exposes the genja module, which wraps the Rust runtime and lets Python code:

• build a runtime from hosts, a full inventory, or a settings file
• run Python-authored tasks
• register Python plugins
• inspect raw and transformed inventory data

Installation

For end users, install the package with pip:

pip install genja-py

The base installation provides the Python library without a genja command. Starting with v0.5.0, install the optional command and terminal UI on a supported platform with:

python -m pip install "genja-py[cli]>=0.5.0,<0.6.0"

The extra selects a compatible genja-cli v0.5.x wheel. The v0.5.0 wheel release covers CPython 3.10–3.14 on Linux x86_64, macOS Apple Silicon and Intel, and Windows x64. Intel macOS has wheels for both packages.

To discover Python tasks, declare their modules in the project's [tool.genja.tasks] configuration, then run genja task list or genja tui. See the installation guide for a first project and the CLI guide for task inspection.

The package currently exposes the genja Python module:

import genja

Quick Start

Create a runtime from a simple host mapping:

import genja
from genja.task import Host, TaskInfo, TaskRuntimeContext, TaskSuccessResult, task


@task(name="backup_config")
class BackupTask:
    def start(
        self,
        task: TaskInfo,
        host: Host,
        context: TaskRuntimeContext,
    ) -> TaskSuccessResult:
        connection = context.connection()
        command_output = None
        if connection is not None:
            command_output = connection.execute_command("show running-config")

        return TaskSuccessResult(
            summary=f"backed up {host.hostname}",
            metadata={"show_running_config": command_output},
        )


genja = genja.Genja.from_hosts({
    "router1": {"hostname": "10.0.0.1", "platform": "ios"},
    "router2": {"hostname": "10.0.0.2", "platform": "nxos"},
}).with_runner("serial")

results = genja.run_task(BackupTask)
print(results.to_dict())

tasks = genja.Tasks()
tasks.add_task(BackupTask)

all_results = genja.run_tasks(tasks)
print([result.task_name for result in all_results])

results.to_dict() returns per-host task results with an outcome payload and separate execution_metadata for host timing and retry attempt information.

TaskRuntimeContext exposes the 1-based retry attempt through context.current_attempt and the resolved connection through context.connection() and context.has_connection(). Execution depth remains internal to the runtime.

For async Python applications, use the async entrypoints:

import asyncio
import genja
from genja.task import Host, TaskInfo, TaskRuntimeContext, TaskSuccessResult, task


@task(name="backup_config_async")
class BackupTaskAsync:
    async def start_async(
        self,
        task: TaskInfo,
        host: Host,
        context: TaskRuntimeContext,
    ) -> TaskSuccessResult:
        connection = context.connection()
        command_output = None
        if connection is not None:
            command_output = await connection.execute_command("show running-config")

        return TaskSuccessResult(
            summary=f"backed up {host.hostname}",
            metadata={"show_running_config": command_output},
        )


async def main() -> None:
    runtime = genja.Genja.from_hosts({
        "router1": {"hostname": "10.0.0.1", "platform": "ios"},
    }).with_runner("serial")

    results = await runtime.run_task_async(BackupTaskAsync)
    print(results.to_dict())


asyncio.run(main())

Use run_task_async(...) and run_tasks_async(...) when composing Genja with asyncio.gather(...) or other async application code. The synchronous run_task(...) and run_tasks(...) entrypoints remain available for scripts and non-async callers.

Python task authoring rules:

• Define def start(...) for blocking tasks.
• Define async def start_async(...) for async tasks.
• Define exactly one of those methods on a @task(...) class.
• Use sub_tasks=[ChildTask, ...] to declare child tasks.
• Use supports_dry_run=True with dry_run(...) or dry_run_async(...) when operators should be able to preview a task.
• Use idempotency=IdempotencyMode.CHECK or idempotency=IdempotencyMode.CHECK_AND_VERIFY with check(...) or check_async(...) when the task can inspect whether a host is already in the desired state.
• Use session_verification=SessionVerificationConfig(...) with connection_plugin_name when a changed task should prove that a new authenticated management session can be established after the change.

from genja.task import (
    Host,
    IdempotencyCheckResult,
    IdempotencyMode,
    TaskInfo,
    TaskRuntimeContext,
    TaskSuccessResult,
    task,
)


@task(name="ensure_ntp", idempotency=IdempotencyMode.CHECK)
class EnsureNtp:
    def check(
        self,
        task: TaskInfo,
        host: Host,
        context: TaskRuntimeContext,
    ) -> IdempotencyCheckResult:
        return IdempotencyCheckResult.change_required(diff="+ntp server 192.0.2.10")

    def start(
        self,
        task: TaskInfo,
        host: Host,
        context: TaskRuntimeContext,
    ) -> TaskSuccessResult:
        return TaskSuccessResult(changed=True, summary="configured NTP")

Dry-run dispatch does not automatically run idempotency checks. Task authors who want shared inspection behavior can call private helper code from both their dry-run hook and check hook.

Session verification is independent from idempotency. When both session_verification=SessionVerificationConfig(...) and idempotency=IdempotencyMode.CHECK_AND_VERIFY are enabled, Genja replaces the connection before running the post-check, so the post-check uses the new session.

Logging

Rust-side runtime logs emitted by Genja are forwarded into Python's standard logging system when the extension module is imported. Configure Python logging handlers and levels before running Genja tasks:

import logging

import genja

logging.basicConfig(level=logging.INFO)

runtime = genja.Genja.from_hosts({
    "router1": {"hostname": "10.0.0.1", "platform": "ios"},
}).with_runner("serial")

Tests can capture those records with pytest's caplog fixture.

Full Inventory

Use genja.inventory when you need groups and defaults:

import genja
from genja.inventory import Defaults, Group, Host, Inventory

inventory = Inventory(
    hosts={
        "router1": Host(hostname="10.0.0.1", groups=["core"]),
    },
    groups={
        "core": Group(platform="ios", data={"role": "core"}),
    },
    defaults=Defaults(username="admin", port=22),
)

genja = genja.Genja.from_inventory(inventory)

print(genja.inventory_full())
print(genja.inventory_raw())

Inventory Accessors

The runtime exposes three inventory views:

• genja.inventory(): raw hosts only
• genja.inventory_full(): transformed hosts, groups, and defaults
• genja.inventory_raw(): raw hosts, groups, and defaults

Plugins

You can register Python plugins directly:

import genja


class MyProcessorPlugin:
    name = "audit"
    group = "ProcessorPlugin"

    def on_task_finish(self, context, results):
        return None


plugins = genja.PluginManager()
plugins.register_plugin(MyProcessorPlugin())

Rust plugins can be loaded from a directory:

plugins = genja.PluginManager()
plugins.load_rust_plugins_from_directory("./plugins")

Settings Files

From A Settings File

Build a runtime from a settings file:

import genja

genja = genja.Genja.from_settings_file("config.yaml")

Settings files are strict: unknown fields fail loading instead of being ignored. Use explicit option maps such as runner.options for plugin-specific values.

From Programmatic Settings

Settings can also be built directly in Python. Use from_settings(...) when inventory should be loaded from settings.inventory:

import genja

settings = genja.Settings(
    inventory=genja.InventoryConfig(
        options=genja.OptionsConfig(
            hosts_file="./inventory/hosts.yaml",
        ),
    ),
    runner=genja.RunnerConfig(
        plugin="serial",
        retry=genja.RunnerRetryConfig(
            allow=True,
            max_attempts=3,
            delay_ms=250,
        ),
    ),
)

genja = genja.Genja.from_settings(settings)

Use from_settings_async(...) for programmatic settings or from_settings_file_async(...) for settings files when the selected inventory plugin is async. Async construction is strict: the selected inventory plugin must be async-capable, and sync-only inventory plugins such as the default FileInventoryPlugin are rejected.

import genja


class ApiInventoryPlugin(genja.InventoryPluginBase):
    name = "api_inventory"

    async def load(self, settings, plugins):
        return {
            "router1": {
                "hostname": "10.0.0.1",
                "platform": "ios",
            },
        }


plugins = genja.PluginManager()
plugins.register_plugin(ApiInventoryPlugin())

settings = genja.Settings(
    inventory=genja.InventoryConfig(plugin="api_inventory"),
)

genja = await genja.Genja.from_settings_async(settings, plugin_manager=plugins)

# Or load settings from a file with the same strict async inventory contract.
genja = await genja.Genja.from_settings_file_async(
    "config.yaml",
    plugin_manager=plugins,
)

Programmatic construction itself does not read files, but runtime creation validates supplied settings before building the runtime. To validate explicitly, call settings.validate() or settings.ssh.validate().

With Explicit Inventory

When host data is supplied directly, from_hosts(...) and from_inventory(...) continue to use that explicit inventory instead of loading from settings.inventory.

With Python Plugins

If you need Python plugins during settings-file loading, provide a plugin manager:

plugins = genja.PluginManager()
genja = genja.Genja.from_settings_file("config.yaml", plugin_manager=plugins)

The same plugin_manager argument is available on Genja.from_settings(...) and the async settings constructors for Python-authored inventory plugins.

Development

The commands below assume a repository checkout and use PDM-managed tooling.

Clone the repository and move into the Python package directory:

git clone git@github.com:Smertan/genja.git
cd genja/genja-core-python

Install the development dependencies:

pdm install -d

Build and install the Rust extension into the project virtual environment:

pdm run maturin develop

Run the Rust-side binding tests:

pdm run test-rust

Use pdm run test-rust instead of plain cargo test. The Rust tests embed Python and need access to the PDM-managed virtualenv packages such as pydantic.

Run the Python test suite:

pdm run test

Run Ruff:

pdm run lint

Check Python API stubs and PyO3 documentation coverage:

pdm run check-stubs

Run this when changing Rust/PyO3-exposed Python APIs, .pyi stubs, Python API docstrings, or top-level Python re-exports.

The check requires every top-level python/genja/*.pyi stub to be listed in STUBS_REQUIRING_DOCSTRINGS in scripts/check_python_api_docs.py. When adding a stub, document its public API and add it to that list during the same change. Add duplicated top-level re-export classes to DUPLICATED_TOP_LEVEL_CLASSES so genja.pyi and __init__.pyi stay aligned.