k3nny/apim-apps
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(apim-apps): first commit

Browse Source
This commit is contained in:
k3nny
2026-06-29 14:22:37 +02:00
commit 7738a51d16
28 changed files with 1973 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apim_apps/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from .client import AxwayClient, _AppendOnlySession
from .engine import normalize_env
from .loader import EXAMPLE_CONFIG, EXAMPLE_MANIFEST, load_config, load_manifest
from .models import Action, ApiAccess, AppSpec, EnvConfig, Manifest, NormResult, OrgSpec
from .report import print_report
apim_apps/_colors.py
+16
View File
@@ -0,0 +1,16 @@
import sys
ANSI = {
"green": "\033[32m",
"yellow": "\033[33m",
"red": "\033[31m",
"cyan": "\033[36m",
"bold": "\033[1m",
"reset": "\033[0m",
}
def _c(color: str, text: str) -> str:
if sys.stdout.isatty():
return f"{ANSI[color]}{text}{ANSI['reset']}"
return text
apim_apps/cli.py
+110
View File
@@ -0,0 +1,110 @@
import argparse
import logging
import os
import sys
import urllib3
from .engine import normalize_env
from .loader import EXAMPLE_CONFIG, EXAMPLE_MANIFEST, load_config, load_manifest
from .models import NormResult
from .report import print_report
from ._colors import _c
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)
def main():
parser = argparse.ArgumentParser(
description="Axway API Manager — normalisation multi-environnements",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--manifest", default="apps.yaml",
help="Manifeste YAML des organisations/applications (défaut: apps.yaml)")
parser.add_argument("--config", default="config.json",
help="Config JSON des environnements Axway (défaut: config.json)")
parser.add_argument("--dry-run", action="store_true",
help="Simule les actions sans appliquer de changements")
parser.add_argument("--env", metavar="ENV",
help="Restreindre l'exécution à cet environnement (ex: --env DEV_LAN)")
parser.add_argument("--init-manifest", action="store_true",
help="Génère un fichier apps.yaml d'exemple et quitte")
parser.add_argument("--init-config", action="store_true",
help="Génère un fichier config.json d'exemple et quitte")
parser.add_argument("--verbose", action="store_true",
help="Affiche les actions skip (debug)")
args = parser.parse_args()
if args.verbose:
logging.getLogger().setLevel(logging.DEBUG)
if args.init_manifest:
with open(args.manifest, "w", encoding="utf-8") as f:
f.write(EXAMPLE_MANIFEST)
print(f"✅ Manifeste d'exemple créé : {args.manifest}")
return
if args.init_config:
with open(args.config, "w", encoding="utf-8") as f:
f.write(EXAMPLE_CONFIG)
print(f"✅ Config d'exemple créée : {args.config}")
return
apim_user = os.environ.get("APIM_USER")
apim_password = os.environ.get("APIM_PASSWORD")
if not apim_user or not apim_password:
log.error("Variables d'environnement APIM_USER et APIM_PASSWORD requises")
sys.exit(1)
for path, label in [(args.manifest, "manifeste"), (args.config, "configuration")]:
if not os.path.exists(path):
log.error(f"Fichier {label} introuvable : {path}")
hint = "--init-manifest" if "yaml" in path else "--init-config"
log.error(f" → Utilisez `{hint}` pour générer un exemple.")
sys.exit(1)
dry_run = args.dry_run
if dry_run:
log.info(_c("yellow", "Mode DRY-RUN activé — aucune modification ne sera effectuée"))
else:
log.warning(_c("red", "⚠️ Mode APPLY — les changements vont être appliqués sur les instances Axway !"))
manifest = load_manifest(args.manifest)
environments = load_config(args.config)
for env in environments:
env.username = apim_user
env.password = apim_password
log.info(f"Manifeste : {len(manifest.organizations)} org(s), {len(manifest.applications)} app(s)")
if args.env:
environments = [e for e in environments if e.name == args.env]
if not environments:
log.error(f"Aucun environnement correspondant à : {args.env!r}")
sys.exit(1)
log.info(f"Environnements ciblés : {[e.name for e in environments]}")
results: dict[str, NormResult] = {}
for env in environments:
try:
results[env.name] = normalize_env(env, manifest, dry_run)
except Exception as e:
r = NormResult()
r.errors.append(f"Erreur fatale : {e}")
results[env.name] = r
log.exception(f"[{env.name}] Erreur fatale non gérée")
print_report(results, dry_run)
total_errors = sum(len(r.errors) for r in results.values())
sys.exit(1 if total_errors else 0)
apim_apps/client.py
+122
View File
@@ -0,0 +1,122 @@
from typing import Optional
import requests
from .models import AppSpec, EnvConfig, OrgSpec
class _AppendOnlySession(requests.Session):
"""HTTP session that refuses DELETE requests — this tool is append-only."""
def request(self, method: str, url, **kwargs):
if method.upper() == "DELETE":
raise RuntimeError(
f"DELETE is not allowed: this tool never removes existing resources ({url})"
)
return super().request(method, url, **kwargs)
class AxwayClient:
"""Axway API Manager REST client. Endpoint: /api/portal/v1.4"""
def __init__(self, env: EnvConfig):
self.env = env
self.base = env.url.rstrip("/") + "/api/portal/v1.4"
self.session = _AppendOnlySession()
self.session.auth = (env.username, env.password)
self.session.verify = env.verify_ssl
self.session.headers.update({"Accept": "application/json"})
# ── Helpers ───────────────────────────────────────────────────────────────
def _get_all(self, path: str, params: dict = None) -> list:
url = f"{self.base}{path}"
p = dict(params or {})
p.setdefault("count", 100)
p["startIndex"] = 0
items = []
while True:
r = self.session.get(url, params=p, timeout=30)
r.raise_for_status()
data = r.json()
if isinstance(data, list):
items.extend(data)
break
chunk = data.get("result", data.get("items", []))
if not isinstance(chunk, list):
return [data]
items.extend(chunk)
total = data.get("total", len(chunk))
p["startIndex"] += len(chunk)
if p["startIndex"] >= total or not chunk:
break
return items
def _post(self, path: str, payload: dict) -> dict:
r = self.session.post(
f"{self.base}{path}",
json=payload,
headers={"Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
return r.json()
# ── Organisations ─────────────────────────────────────────────────────────
def list_orgs(self) -> dict[str, dict]:
return {o["name"]: o for o in self._get_all("/organizations")}
def create_org(self, spec: OrgSpec) -> dict:
return self._post("/organizations", {
"name": spec.name,
"description": spec.description,
"email": spec.email,
"phone": spec.phone,
"enabled": spec.enabled,
"development": spec.development,
})
# ── Applications ──────────────────────────────────────────────────────────
def list_apps(self) -> dict[tuple, dict]:
apps = self._get_all("/applications")
orgs_by_id = {o["id"]: o["name"] for o in self._get_all("/organizations")}
return {
(orgs_by_id.get(a.get("organizationId", ""), "?"), a["name"]): a
for a in apps
}
def create_app(self, spec: AppSpec, org_id: str) -> dict:
return self._post("/applications", {
"name": spec.name,
"organizationId": org_id,
"description": spec.description,
"email": spec.email,
"phone": spec.phone,
"enabled": spec.enabled,
"state": "approved",
})
# ── APIs front-end ────────────────────────────────────────────────────────
def list_frontend_apis(self) -> list[dict]:
return self._get_all("/proxies")
def find_frontend_api(self, api_name: str, api_version: str = "") -> Optional[dict]:
for api in self.list_frontend_apis():
if api.get("name", "") == api_name:
if not api_version or api.get("version", "") == api_version:
return api
return None
# ── Souscriptions ─────────────────────────────────────────────────────────
def list_app_api_access(self, app_id: str) -> list[dict]:
try:
return self._get_all(f"/applications/{app_id}/apis")
except requests.HTTPError:
return []
def grant_api_access(self, app_id: str, api_id: str) -> dict:
return self._post(f"/applications/{app_id}/apis/{api_id}", {})
apim_apps/engine.py
+173
View File
@@ -0,0 +1,173 @@
import logging
from ._colors import _c
from .client import AxwayClient
from .models import Action, EnvConfig, Manifest, NormResult
log = logging.getLogger(__name__)
def _applies_to(target_envs: list[str], env_name: str) -> bool:
return not target_envs or env_name in target_envs
def normalize_env(env: EnvConfig, manifest: Manifest, dry_run: bool) -> NormResult:
result = NormResult()
tag = f"[{env.name}]"
log.info(f"{tag} Connexion à {env.url} …")
client = AxwayClient(env)
# ── 1. Organisations ──────────────────────────────────────────────────────
try:
existing_orgs = client.list_orgs()
except Exception as e:
result.errors.append(f"{tag} Impossible de lister les organisations : {e}")
return result
for org_spec in manifest.organizations:
if not _applies_to(org_spec.environments, env.name):
continue
action = Action(
kind="create_org",
env_name=env.name,
description=f"Créer organisation « {org_spec.name} »",
detail={"org": org_spec.name},
)
if org_spec.name in existing_orgs:
log.debug(f"{tag} Org « {org_spec.name} » déjà présente — skip")
result.actions_skipped.append(action)
continue
log.info(_c("yellow", f"{tag} {'[DRY-RUN] ' if dry_run else ''}Création org « {org_spec.name} »"))
if not dry_run:
try:
created = client.create_org(org_spec)
existing_orgs[org_spec.name] = created
log.info(_c("green", f"{tag} ✅ Org « {org_spec.name} » créée (id={created.get('id')})"))
except Exception as e:
msg = f"{tag} ❌ Échec création org « {org_spec.name} » : {e}"
log.error(msg)
result.errors.append(msg)
continue
else:
# Placeholder so apps referencing this org are processed in dry-run
existing_orgs[org_spec.name] = {"id": "", "name": org_spec.name}
result.actions_done.append(action)
# ── 2. Applications ───────────────────────────────────────────────────────
try:
existing_apps = client.list_apps()
except Exception as e:
result.errors.append(f"{tag} Impossible de lister les applications : {e}")
return result
for app_spec in manifest.applications:
if not _applies_to(app_spec.environments, env.name):
continue
key = (app_spec.org_name, app_spec.name)
action_app = Action(
kind="create_app",
env_name=env.name,
description=f"Créer application « {app_spec.name} » dans org « {app_spec.org_name} »",
detail={"org": app_spec.org_name, "app": app_spec.name},
)
raw_org = existing_orgs.get(app_spec.org_name)
if raw_org is None:
msg = (
f"{tag} ❌ Organisation « {app_spec.org_name} » introuvable dans l'API Manager "
f"— application « {app_spec.name} » ignorée. "
f"Créez l'organisation manuellement dans l'API Manager."
)
log.error(msg)
result.errors.append(msg)
continue
org_id = raw_org["id"]
if key not in existing_apps:
log.info(_c("yellow", f"{tag} {'[DRY-RUN] ' if dry_run else ''}Création app « {app_spec.name} » (org={app_spec.org_name})"))
if not dry_run:
try:
created_app = client.create_app(app_spec, org_id)
existing_apps[key] = created_app
log.info(_c("green", f"{tag} ✅ App « {app_spec.name} » créée (id={created_app.get('id')})"))
except Exception as e:
msg = f"{tag} ❌ Échec création app « {app_spec.name} » : {e}"
log.error(msg)
result.errors.append(msg)
continue
result.actions_done.append(action_app)
else:
result.actions_skipped.append(action_app)
# ── 3. Souscriptions aux APIs ─────────────────────────────────────────
if not app_spec.apis:
continue
raw_app = existing_apps.get(key)
if raw_app is None:
continue # app non créée (dry-run ou erreur)
app_id = raw_app.get("id", "")
try:
granted = client.list_app_api_access(app_id)
granted_ids = {a.get("apiId", a.get("id", "")) for a in granted}
granted_names = {a.get("apiName", a.get("name", "")) for a in granted}
except Exception as e:
log.warning(f"{tag} Impossible de lire les accès API de « {app_spec.name} » : {e}")
granted_ids = set()
granted_names = set()
for api_access in app_spec.apis:
action_api = Action(
kind="grant_api",
env_name=env.name,
description=(
f"Abonner « {app_spec.name} » à l'API « {api_access.api_name} »"
+ (f" v{api_access.api_version}" if api_access.api_version else "")
),
detail={"app": app_spec.name, "api": api_access.api_name, "version": api_access.api_version},
)
if api_access.api_name in granted_names:
log.debug(f"{tag} API « {api_access.api_name} » déjà accordée à « {app_spec.name} » — skip")
result.actions_skipped.append(action_api)
continue
if dry_run:
log.info(_c("yellow", f"{tag} [DRY-RUN] Abonnement « {app_spec.name} » → API « {api_access.api_name} »"))
result.actions_done.append(action_api)
continue
fe_api = client.find_frontend_api(api_access.api_name, api_access.api_version)
if fe_api is None:
msg = (
f"{tag} ⚠️ API front-end « {api_access.api_name} » introuvable "
f"— souscription de « {app_spec.name} » ignorée"
)
log.warning(msg)
result.errors.append(msg)
continue
api_id = fe_api["id"]
if api_id in granted_ids:
result.actions_skipped.append(action_api)
continue
try:
client.grant_api_access(app_id, api_id)
log.info(_c("green", f"{tag} ✅ Accès API « {api_access.api_name} » accordé à « {app_spec.name} »"))
result.actions_done.append(action_api)
except Exception as e:
msg = f"{tag} ❌ Échec souscription « {app_spec.name} » → « {api_access.api_name} » : {e}"
log.error(msg)
result.errors.append(msg)
return result
apim_apps/loader.py
+125
View File
@@ -0,0 +1,125 @@
import json
import textwrap
import yaml
from .models import ApiAccess, AppSpec, EnvConfig, Manifest, OrgSpec
EXAMPLE_MANIFEST = textwrap.dedent("""\
# =============================================================================
# Manifeste de normalisation Axway API Manager
# =============================================================================
# Décrit les applications clientes à garantir sur les environnements.
#
# Les organisations sont gérées manuellement dans l'API Manager.
# Ce manifeste ne contient que la section "applications".
#
# Chaque objet peut cibler des environnements spécifiques via "environments:".
# Si "environments" est absent ou vide → s'applique à TOUS les environnements.
# =============================================================================
applications:
# Application présente sur tous les environnements
- name: "MonAppMobile"
organization: "Partenaires-Externes"
description: "Application mobile grand public"
email: "mobile-app@example.com"
enabled: true
apis:
- name: "Catalogue-Produits"
version: "v2"
- name: "Commandes"
version: "v1"
- "Notifications" # syntaxe courte (sans version)
# Application restreinte à PROD et PREPROD
- name: "BackOffice-CRM"
organization: "Equipe-Integration"
description: "Connecteur CRM back-office"
email: "crm-team@example.com"
enabled: true
environments:
- PREPROD
- PROD
apis:
- name: "Clients"
version: "v3"
- name: "Contrats"
version: "v1"
# Application de test uniquement sur DEV et RECETTE
- name: "Robot-Tests-API"
organization: "Equipe-QA"
description: "Client Robot Framework pour tests d'intégration"
enabled: true
environments:
- DEV
- RECETTE
apis:
- "Catalogue-Produits"
- "Commandes"
- "Clients"
""")
EXAMPLE_CONFIG = json.dumps({
"environments": [
{"name": "DEV", "url": "https://apimgr-dev.example.com:8075", "verify_ssl": False},
{"name": "RECETTE", "url": "https://apimgr-rec.example.com:8075", "verify_ssl": False},
{"name": "PREPROD", "url": "https://apimgr-ppd.example.com:8075", "verify_ssl": False},
{"name": "PROD", "url": "https://apimgr-prd.example.com:8075", "verify_ssl": True},
]
}, indent=2, ensure_ascii=False)
def load_manifest(path: str) -> Manifest:
with open(path, encoding="utf-8") as f:
raw = yaml.safe_load(f)
manifest = Manifest()
for o in raw.get("organizations", []):
manifest.organizations.append(OrgSpec(
name=o["name"],
description=o.get("description", ""),
email=o.get("email", ""),
phone=o.get("phone", ""),
enabled=o.get("enabled", True),
development=o.get("development", False),
environments=o.get("environments", []),
))
for a in raw.get("applications", []):
apis = []
for api in a.get("apis", []):
if isinstance(api, str):
apis.append(ApiAccess(api_name=api))
else:
apis.append(ApiAccess(api_name=api["name"], api_version=api.get("version", "")))
manifest.applications.append(AppSpec(
name=a["name"],
org_name=a["organization"],
description=a.get("description", ""),
email=a.get("email", ""),
phone=a.get("phone", ""),
enabled=a.get("enabled", True),
apis=apis,
environments=a.get("environments", []),
))
return manifest
def load_config(path: str) -> list[EnvConfig]:
with open(path, encoding="utf-8") as f:
raw = json.load(f)
return [
EnvConfig(
name=e["name"],
url=e["url"],
username="",
password="",
verify_ssl=e.get("verify_ssl", False),
)
for e in raw["environments"]
]
apim_apps/models.py
+60
View File
@@ -0,0 +1,60 @@
from dataclasses import dataclass, field
@dataclass
class ApiAccess:
api_name: str
api_version: str = ""
@dataclass
class AppSpec:
name: str
org_name: str
description: str = ""
email: str = ""
phone: str = ""
enabled: bool = True
apis: list["ApiAccess"] = field(default_factory=list)
environments: list[str] = field(default_factory=list)
@dataclass
class OrgSpec:
name: str
description: str = ""
email: str = ""
phone: str = ""
enabled: bool = True
development: bool = False
environments: list[str] = field(default_factory=list)
@dataclass
class Manifest:
organizations: list[OrgSpec] = field(default_factory=list)
applications: list[AppSpec] = field(default_factory=list)
@dataclass
class EnvConfig:
name: str
url: str
username: str
password: str
verify_ssl: bool = False
@dataclass
class Action:
kind: str # "create_org" | "create_app" | "grant_api"
env_name: str
description: str
detail: dict = field(default_factory=dict)
@dataclass
class NormResult:
actions_done: list[Action] = field(default_factory=list)
actions_skipped: list[Action] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
apim_apps/report.py
+52
View File
@@ -0,0 +1,52 @@
from ._colors import _c
from .models import NormResult
def print_report(results: dict[str, NormResult], dry_run: bool):
mode = _c("yellow", "[DRY-RUN] ") if dry_run else ""
print()
print(_c("bold", f"{'─'*65}"))
print(_c("bold", f" {mode}Rapport de normalisation"))
print(_c("bold", f"{'─'*65}"))
total_done = total_skip = total_err = 0
for env_name, result in results.items():
done = len(result.actions_done)
skip = len(result.actions_skipped)
err = len(result.errors)
total_done += done
total_skip += skip
total_err += err
status = _c("green", "✅") if err == 0 else _c("red", "❌")
print(f"\n {status} {_c('bold', env_name)}")
print(f" Actions {'simulées' if dry_run else 'effectuées'} : {_c('green', str(done))}")
print(f" Déjà présents (skip) : {_c('cyan', str(skip))}")
print(f" Erreurs / avertissements : {_c('red', str(err)) if err else _c('green', '0')}")
if result.actions_done:
print(f" {'─'*50}")
for a in result.actions_done:
icon = {"create_org": "🏢", "create_app": "📱", "grant_api": "🔗"}.get(a.kind, "•")
prefix = " ~ " if dry_run else " + "
print(f" {prefix}{icon} {a.description}")
if result.errors:
print(f" {'─'*50}")
for e in result.errors:
print(f" ⚠️ {_c('red', e)}")
print()
print(_c("bold", f"{'─'*65}"))
print(
f" Total — {'simulé' if dry_run else 'effectué'}: {_c('green', str(total_done))} | "
f"skip: {_c('cyan', str(total_skip))} | "
f"erreurs: {_c('red', str(total_err)) if total_err else _c('green', '0')}"
)
print(_c("bold", f"{'─'*65}"))
if dry_run:
print(_c("yellow", "\n ℹ️ Mode DRY-RUN : aucun changement appliqué."))
print(_c("yellow", " Relancez sans --dry-run pour appliquer les modifications.\n"))
else:
print()