python-rundeck 0.2.0

Python client for the Rundeck API (v14-v56).

python-rundeck

Client Python pour l’API Rundeck (v14–v56), inspiré de l’architecture de python-gitlab. Fournit des managers typés pour les ressources clés (projects, jobs, executions, tokens, users, système, configuration) et SCM (import/export).

Sommaire

• Installation
• Démarrage rapide
• Configuration
• Ressources disponibles
• Exemples par ressource
• Gestion des erreurs
• Développement et tests

Installation

Prerequis : Python 3.11+. Le projet utilise Poetry.

poetry install
# ou en editable avec pip
pip install -e .

Démarrage rapide

from rundeck.client import Rundeck

rd = Rundeck(url="https://rundeck.example.com", token="MY_TOKEN", api_version=56)
# Auth par mot de passe (session cookie) si pas de token
rd = Rundeck(url="https://rundeck.example.com", username="admin", password="admin", api_version=56)

# Lister les projets
projects = rd.projects.list()
for p in projects:
    print(p.name)

# Récupérer un projet et lancer un job
project = rd.projects.get("demo")
jobs = project.jobs.list()
execu = jobs[0].run()
print(execu.id)

Configuration

La configuration suit un modèle en cascade (args > env > fichiers > défauts) via RundeckConfig.

Paramètres principaux :

• url : URL Rundeck (ex. https://rundeck.example.com)
• token : Token API (en-tête X-Rundeck-Auth-Token)
• username / password : Authentification par session (j_security_check) si aucun token n'est fourni.
• api_version : Version d’API (ex. 56)
• timeout : Timeout des requêtes (float, secondes)
• ssl_verify : Vérification TLS (bool ou chemin CA)

Fichiers de config : si besoin, passez config_files ou Rundeck.from_config(config_section=...). Variables d'env utiles (config client) :

• RUNDECK_URL : URL de base
• RUNDECK_TOKEN : token API (en-tête X-Rundeck-Auth-Token)
• RUNDECK_USERNAME / RUNDECK_PASSWORD : auth session (si pas de token)
• RUNDECK_API_VERSION : version d'API (ex: 56)
• RUNDECK_TIMEOUT : timeout des requêtes (secondes)
• RUNDECK_SSL_VERIFY : vérification TLS (bool ou chemin CA)
• RUNDECK_USER_AGENT : User-Agent HTTP

Ressources disponibles

• projects (ProjectManager) : CRUD projets, export/import de jobs, config projet, archive (export/import).
• jobs (JobManager) : liste, get, suppressions, actions bulk, exécution.
• executions (ExecutionManager) : liste/filtre, running, get/delete, query avancée.
• tokens (TokenManager) : liste, get, create, delete.
• users (UserManager) : opérations utilisateurs (selon implémentation courante).
• metrics (MetricsManager) : endpoints /metrics (list/data/healthcheck/ping).
• plugins (PluginManager) : liste des plugins installés (/plugin/list).
• webhooks (WebhookEventManager + ProjectWebhookManager) : gestion des webhooks projet et envoi via token.
• key_storage (StorageKeyManager) : gestion du stockage des clés /storage/keys.
• adhoc (AdhocManager, via project.adhoc) : exécution de commandes/scripts AdHoc.
• system (SystemManager) : info système, exécutions enable/disable, logstorage, ACL.
• config_management (ConfigManagementManager) : configuration globale /config.
• scm (via project.scm et job.scm) : plugins import/export, setup, enable/disable, statut, actions (commit/import/export...).

Exemples par ressource (complets)

Projets

# CRUD projet
p = rd.projects.create("demo")
p = rd.projects.get("demo")
rd.projects.delete("demo")
projects = rd.projects.list()

# Export / import de jobs d'un projet (via le manager de jobs)
project.jobs.export(format="json", idlist="id1,id2", groupPath="group/sub")
project.jobs.import_jobs(content=open("jobs.json").read(), format="json", dupeOption="update", uuidOption="remove")

# Export / import du projet (archive ZIP)
archive = project.archive
resp = archive.export(export_all=False, export_webhooks=True)  # Response brute (zip)
token_info = archive.export_async(exportAll=True)
status = archive.export_status(token_info.get("token", ""))
zip_resp = archive.export_download(token_info.get("token", ""))

# Import d'archive (synchrone ou async)
archive.import_archive(
    content=open("project-export.zip", "rb").read(),
    jobUuidOption="preserve",
    importExecutions=True,
    importConfig=True,
)
archive.import_archive(content=open("project-export.zip", "rb").read(), async_import=True)
archive.import_status()

# Readme / MOTD du projet
project.readme.get_readme()  # texte par défaut
project.readme.get_readme(accept="application/json")
project.readme.update_readme("Nouveau contenu", content_type="text/plain")
project.readme.delete_readme()
project.readme.get_motd()
project.readme.update_motd("Message du jour", content_type="text/plain")
project.readme.delete_motd()

# Config projet (clé/valeur)
conf = p.config.get()
p.config.keys.get("project.label")
p.config.keys.set("project.label", "Demo")
p.config.keys.update({"project.description": "Sample"})
p.config.replace({"project.label": "Demo", "project.description": "Sample"})
p.config.keys.delete("project.label")

# SCM import/export (sur un projet)
scm_import = project.scm.import_  # ou getattr(project.scm, "import")
scm_export = project.scm.export

# Découverte des plugins
scm_import.plugins.list()
scm_export.plugins.list()

# Champs d'entrée pour un plugin et setup (plugin_type explicite)
fields = scm_import.plugins.input_fields("git-import")
scm_import.config.setup("git-import", {"url": "ssh://[e-mail]/repo.git", "dir": "/tmp/repo"})

# Activer/désactiver un plugin
scm_import.config.enable("git-import")
scm_export.config.disable("custom-export")

# Statut/config SCM
import_status = scm_import.actions.status()
export_conf = scm_export.config.get()

# Actions SCM côté projet (ex: commit/pull/push selon plugin)
action_fields = scm_export.actions.input_fields("commit")
scm_export.actions.perform(
    "commit",
    input_values={"message": "Sync jobs"},
    jobs=["job-1"],
    items=["path/job-1.yaml"],
    deleted=["obsolete/path.yaml"],
)

Jobs

# Lister les jobs d’un projet
jobs = rd.jobs.list(project="demo", groupPath="ops")

# Depuis un projet parenté
project = rd.projects.get("demo")
jobs = project.jobs.list()

# Accès direct à un job
job = rd.jobs.get("job-id")
job.delete()
job.definition(format="yaml")
job.retry("exec-id", argString="-opt val")
job.enable_execution()
job.disable_execution()
job.enable_schedule()
job.disable_schedule()
info = job.info()
meta = job.meta(meta="name,description")
tags = job.tags()
workflow = job.workflow()
forecast = job.forecast(time="2024-05-01T10:00:00Z", max=5)

# Exporter/importer des jobs via le manager (paramètre project ou parent)
rd.jobs.export(project="demo", format="xml", idlist="id1,id2", groupPath="group")
rd.jobs.import_jobs(
    project="demo",
    content=open("jobs.xml", "rb").read(),
    fileformat="xml",
    dupeOption="update",
)
# Ou via un projet parenté
project.jobs.export(format="json")
project.jobs.import_jobs(content=open("jobs.json", "rb").read(), fileformat="json")

Note: l'import reste exposé sous le nom `import_jobs(...)` (le mot-clé Python empêche un appel direct à `.import`). Si vous préférez l'alias, utilisez `getattr(rd.jobs, "import")(...)`.

# Exécuter et récupérer l'exécution
execution = job.run(argString="-option value")

# Actions bulk
rd.jobs.bulk.enable_execution(["id1", "id2"])
rd.jobs.bulk.disable_execution(["id1", "id2"])
rd.jobs.bulk.delete(["id1", "id2"])
rd.jobs.bulk.enable_schedule(["id1", "id2"])
rd.jobs.bulk.disable_schedule(["id1", "id2"])

# Upload de fichiers d'option et fichiers uploadés
job.upload_option_file("csvfile", open("data.csv", "rb").read(), file_name="data.csv")
job.list_uploaded_files(max=20)
rd.jobs.get_uploaded_file_info("file-id")

# SCM import/export sur un job
job_scm_export = job.scm.export
job_scm_import = job.scm.import_  # ou getattr(job.scm, "import")

job_scm_export.status()
job_scm_export.diff()
job_scm_export.perform("commit", input_values={"message": "Sync job"})

job_scm_import.status()
job_scm_import.input_fields("pull")
job_scm_import.perform("pull", input_values={"message": "Update from repo"})

# Ressources d'un projet
resources = project.resources.list(format="json", groupPath="ops")
node = project.resources.get("node1")
sources = project.sources.list()
source_details = project.sources.get(1)
project.sources.list_resources(1, accept="application/json")
project.sources.update_resources(1, content="{}", content_type="application/json")
project.acl.list()
project.acl.get("policy.aclpolicy")
project.acl.create("policy.aclpolicy", content="...yaml...")
project.acl.update("policy.aclpolicy", content="...yaml...")
project.acl.delete("policy.aclpolicy")

Exécutions

# Liste simple ou paginée
execs = rd.executions.list(project="demo", status="running", max=50, offset=0)
running = rd.executions.running(project="demo")  # ou "*" pour tous

# Détails / suppression
e = rd.executions.get("123")
rd.executions.delete("123")
rd.executions.bulk_delete(["123", "124"])

# Requête avancée (query)
advanced = rd.executions.query(
    project="demo",
    statusFilter="failed",
    userFilter="alice",
    jobIdListFilter=["id1", "id2"],
    groupPath="ops",
    max=100,
)

# Méthodes sur Execution
e.abort(asUser="admin")
output = e.get_output(offset=0, maxlines=100)
state = e.get_state()
is_running = e.is_running()
e.refresh()  # recharge les données

Tokens

tokens = rd.tokens.list()
user_tokens = rd.tokens.list(user="alice")
t = rd.tokens.get("tok-1")
new_token = rd.tokens.create(user="alice", roles=["admin"], duration="90d", name="cli")
rd.tokens.delete(new_token.id)

Utilisateurs

users = rd.users.list()
me = rd.users.get_current()
u = rd.users.get("bob")
roles = rd.users.current_roles()

# Mise à jour via manager ou objet
u = rd.users.update("bob", firstName="Bob", lastName="Builder", email="[e-mail]")
u.roles()            # via l'objet
u.update(email="[e-mail]")  # met à jour et rafraîchit l'objet

Système

system = rd.system
info = system.info()

# Log storage
system.logstorage.info()
system.logstorage.incomplete(max=50, offset=0)
system.logstorage.incomplete_resume()

# Mode exécution (sous-manager)
system.executions.enable()
system.executions.disable()
system.executions.status()

# ACL
system.acl.list()
system.acl.get("policy.aclpolicy")
system.acl.create("policy.aclpolicy", content="...yaml...")
system.acl.update("policy.aclpolicy", content="...yaml...")
system.acl.delete("policy.aclpolicy")

# Scheduler takeover (cluster)
rd.scheduler.takeover(all_servers=True)
rd.scheduler.takeover(server_uuid="uuid-123", project="demo", job_id="job-1")

Métriques

metrics = rd.metrics
metrics.list()
metrics.data()
metrics.healthcheck()
metrics.ping()

# Plugins installés
plugins = rd.plugins.list()
for plugin in plugins:
    print(plugin.name, plugin.service)

# Détail d'un plugin
first = plugins[0]
detail = rd.plugins.detail(first.service, first.name)

# Webhooks projet et envoi
project = rd.projects.get("demo")
wh = project.webhooks
wh.create(
    project=project.id,
    name="hook1",
    user="admin",
    roles="admin",
    eventPlugin="log-webhook-event",
    config={},
    enabled=True,
)
hooks = wh.list()
first_hook = hooks[0]
wh.update(first_hook.id, name="hook1-updated")
rd.webhooks.send(first_hook.authToken, json={"hello": "world"})
wh.delete(first_hook.id)

# Key storage (/storage/keys)
ks = rd.key_storage
# Créer un secret (password)
ks.create("integration/secret1", content="s3cr3t", content_type="application/x-rundeck-data-password")
resources = ks.list()  # liste racine
meta = ks.get_metadata("integration/secret1")
ks.delete("integration/secret1")

# AdHoc commands/scripts
project = rd.projects.get("demo")
exec1 = project.adhoc.run_command("echo 'hello world'")
exec2 = project.adhoc.run_script("echo 'from script'")
# exec2.id pour suivre l'exécution
# Mode stub (sans refresh immédiat) puis refresh manuel
exec_stub = project.adhoc.run_command("echo stub", refresh=False)
exec_stub.refresh()  # charge l'objet complet
# Script via multipart (upload)
exec3 = project.adhoc.run_script(
    script_file=("hello.sh", "echo multipart", "text/plain"),
    refresh=False,
)

# Features système
features = rd.features.list()
if features:
    first = features[0]
    status = rd.features.get(first.name)
    print(first.name, status.enabled)

Configuration globale /config

cfg = rd.config_management
all_configs = cfg.list()
cfg.save([{"key": "ui.banner", "value": "Hello"}])
cfg.delete("ui.banner", strata="default")
cfg.refresh()
cfg.restart()

Pagination

Les managers héritent de RundeckObjectManager.iter(...) (offset/max). Exemple :

for job in rd.jobs.iter(project="demo", page_size=100):
    print(job.id)

Gestion des erreurs

Les erreurs HTTP passent par raise_for_status et lèvent des exceptions dédiées (ex. RundeckAuthenticationError, RundeckNotFoundError, RundeckValidationError, RundeckConflictError, RundeckServerError). Gérez-les avec un bloc try/except autour des appels client.

Développement et tests

• Formatage/Lint : black src/rundeck tests/, ruff check src/rundeck tests/
• Typage : mypy src/rundeck/
• Tests : pytest
• Tests d'intégration locaux : scripts/run-integration.sh (démarre docker compose à la racine, exporte par défaut RUNDECK_URL=http://localhost:4440, RUNDECK_TOKEN=adminToken, RUNDECK_API_VERSION=56, attend le healthcheck, lance poetry run pytest -m integration). Vous pouvez surcharger les variables d'environnement avant l'appel. Ajoutez KEEP_STACK=1 pour ne pas arrêter l'instance après les tests. Script de teardown dédié : scripts/stop-integration.sh.

Structure du code

• src/rundeck/base.py : objets/manager génériques, helpers _build_path, pagination, CRUD.
• src/rundeck/client.py : client HTTP, méthodes http_get/post/put/delete/list.
• src/rundeck/v1/objects/ : managers et objets métiers (projects, jobs, executions, system, tokens, users, config_management).
• Schéma managers/objets : on applique le pattern manager → objet partout où l’API expose une ressource identifiable (jobs, projects, users, tokens, executions, etc.). Pour les endpoints purement globaux ou utilitaires sans ressource (ex. /system et ses sous-domaines, /config, /metrics), on utilise des sous-managers dédiés plutôt que de forcer un objet artificiel.