id: python-release-semver-gate
namespace: company.team
description: |
Python Release SemVer Gate. Answers one question before a release is tagged: does the proposed version of a Python
package match the public API changes between a base commit and a candidate commit, under a strict Semantic
Versioning policy? Both commits are analysed statically; repository code is never imported, installed, built or run.
inputs:
- id: repo_url
type: STRING
displayName: GitHub repository
description: Public GitHub repository, https://github.com/<owner>/<repo>. No
other host, scheme, port or credentials.
defaults: https://github.com/AyushVUpadhye/kestra-python-semver-gate-demo
validator: '^https://github\.com/[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?/[A-Za-z0-9_][A-Za-z0-9._-]{0,99}$'
- id: base_commit
type: STRING
displayName: Base commit SHA
description: Full 40-character SHA of the last release. Branch and tag names are
not accepted.
defaults: cbe742eeb13641ad1bf66a5403b60a6e7d8e94b4
validator: '^[0-9a-f]{40}$'
- id: candidate_commit
type: STRING
displayName: Candidate commit SHA
description: Full 40-character SHA of the release candidate. Branch and tag
names are not accepted.
defaults: f3359619076dc692e700e1f0a504e8282e644012
validator: '^[0-9a-f]{40}$'
- id: package
type: STRING
displayName: Import name
description: Top-level import name of the package, for example requests. Not the
distribution name on PyPI.
defaults: semver_demo
validator: '^[A-Za-z_][A-Za-z0-9_]{0,99}$'
- id: project_path
type: STRING
displayName: Project path
description: Folder holding pyproject.toml, relative to the repository root, or
"." for the root. The package is looked up in <project_path>/src/<package>
and <project_path>/<package>, separately for each commit.
defaults: "."
validator: '^(?:\.|(?!.*(?:^|/)\.\.?(?:/|$))[A-Za-z0-9._][A-Za-z0-9._-]{0,99}(?:/[A-Za-z0-9._][A-Za-z0-9._-]{0,99}){0,19})$'
- id: base_version
type: STRING
displayName: Base version
description: Optional. Version of the base commit, X.Y.Z. When empty or not set,
[project].version is read from the base pyproject.toml.
required: false
validator: '^(?:(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8}))?$'
- id: proposed_version
type: STRING
displayName: Proposed version
description: Optional. Version proposed for the candidate, X.Y.Z. When empty or
not set, [project].version is read from the candidate pyproject.toml.
required: false
validator: '^(?:(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8}))?$'
- id: pr_number
type: INT
displayName: Pull request number
description: Optional. When set, the verdict is posted on this pull request of
the same repository, and kept up to date on later runs.
required: false
min: 1
tasks:
- id: gate
type: io.kestra.plugin.core.flow.WorkingDirectory
description: One workspace for the pinned analysis tools and the two checkouts.
inputFiles:
requirements.txt: |
griffelib==2.3.0 --hash=sha256:1b8f9cd525681c26b1d6d574faa1371651e8459ca51d209684f50b8096ae06e0
packaging==25.0 --hash=sha256:29572ef2b1f17581046b3a2227d5c611fb25ec70ca1ba8554b24b0e69331a484
gate.py: |
import ast
import enum
import json
import logging
import os
import re
import sys
sys.path.insert(0, "/opt/gate")
FINAL = re.compile(r"(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8})")
REMOVED = {"Public object was removed", "Public re-export was removed"}
INFORMATIONAL = {"Attribute value was changed"}
SCHEMA = "kestra-python-release-semver-gate/1.0"
SIDES = ("base", "candidate")
def describe(value):
"""Readable text for a Griffe value: source text for expressions, paths for objects, name: annotation = default for parameters."""
if value is None:
return None
if isinstance(value, (list, tuple)):
return ", ".join(describe(v) or "" for v in value)
if isinstance(value, enum.Enum):
return str(value.value)
if hasattr(value, "annotation") and hasattr(value, "default") and hasattr(value, "kind"):
text = value.name
if value.annotation is not None:
text += ": " + str(value.annotation)
if value.default is not None:
text += " = " + str(value.default)
return text
if hasattr(value, "members") and hasattr(value, "path"):
return value.path
return str(value)
def short(value, limit=200):
text = "" if value is None else str(value)
return text if len(text) <= limit else text[:limit - 3] + "..."
def detect_root(clone, project, package):
"""Package root for one commit: <project>/src or <project>. Nothing else is searched."""
top = os.path.realpath(clone)
for sub in ("src", ""):
root = os.path.realpath(os.path.join(clone, project, sub))
if root != top and not root.startswith(top + os.sep):
continue
if os.path.isfile(os.path.join(root, package, "__init__.py")) or os.path.isfile(os.path.join(root, package + ".py")):
return root
return None
def unparsable(root, package):
"""First package file Python cannot parse. Griffe skips such modules silently, which would look like removed API."""
single = os.path.join(root, package + ".py")
if os.path.isfile(single):
files = [single]
else:
files = []
for folder, dirs, names in os.walk(os.path.join(root, package)):
dirs[:] = sorted(d for d in dirs if os.path.isfile(os.path.join(folder, d, "__init__.py")))
files += [os.path.join(folder, n) for n in sorted(names) if n.endswith(".py")]
for path in files:
try:
with open(path, "rb") as f:
ast.parse(f.read(), filename=path)
except SyntaxError as exc:
return os.path.relpath(path, root).replace(os.sep, "/") + ":" + str(exc.lineno or 0)
except (ValueError, RecursionError, MemoryError):
return os.path.relpath(path, root).replace(os.sep, "/")
return None
def read_version(clone, project):
"""[project].version from the project's pyproject.toml, read as data. Returns (version, note)."""
import tomllib
path = os.path.join(clone, project, "pyproject.toml")
if not os.path.isfile(path):
return None, "no pyproject.toml"
try:
with open(path, "rb") as f:
data = tomllib.load(f)
except (tomllib.TOMLDecodeError, UnicodeDecodeError, OSError):
return None, "pyproject.toml is not valid TOML"
table = data.get("project")
if not isinstance(table, dict):
return None, "no [project] table"
if "version" in (table.get("dynamic") or []):
return None, "version is dynamic"
version = table.get("version")
if not isinstance(version, str):
return None, "no static [project].version"
if not FINAL.fullmatch(version):
return None, "[project].version " + short(version, 40) + " is not a final X.Y.Z release"
return version, "pyproject.toml"
def modules(obj):
yield obj
for member in obj.members.values():
if not member.is_alias and member.is_module:
yield from modules(member)
def exports_of(obj):
exports = getattr(obj, "exports", None) if obj.is_module else None
return None if exports is None else {str(e) for e in exports}
def is_public(parent, name, member, package):
exports = exports_of(parent)
if member.is_alias:
if not parent.is_module:
return False
if exports is not None:
return name in exports
# A name imported into a package namespace (__init__.py) that points inside the package is a re-export.
target = member.target_path
return parent.is_init_module and not name.startswith("_") and (target == package or target.startswith(package + "."))
if name.startswith("__") and name.endswith("__") and len(name) > 4:
return parent.is_class and member.is_function
return member.is_public
def resolve(alias, package):
"""Target of a re-export, looked up in the already loaded tree only. Never imports anything."""
try:
target = alias.final_target
except Exception:
return None
if target.is_alias or not (target.path == package or target.path.startswith(package + ".")):
return None
return target
def surface(obj, package, out, targets, prefix=None, seen=()):
"""Public API as {path: {kind, parameters, target}}. Re-exports are described by what they point to."""
prefix = prefix or obj.path
for name in sorted(obj.members):
member = obj.members[name]
if not is_public(obj, name, member, package):
continue
path = prefix + "." + name
entry, target = {}, member
if member.is_alias:
entry["target"] = member.target_path
target = resolve(member, package)
if target is None:
out[path] = dict(entry, kind="re-export")
continue
if target.is_module and target.path != path:
# "import pkg" or "from pkg.utils import helpers" inside a package is an import, not API. A module
# listed in __all__ is recorded by name only, so the same API is not counted twice.
if exports_of(obj) is not None:
out[path] = dict(entry, kind="re-export")
continue
if target.path in seen:
continue
if target.path != path:
targets[path] = target
entry["kind"] = target.kind.value
if target.is_function:
entry["parameters"] = [[p.name, p.kind.value, p.default is not None] for p in target.parameters]
out[path] = entry
if target.is_module or target.is_class:
surface(target, package, out, targets, path, seen + (target.path,))
return out
def relative(path, roots):
for side, root in roots.items():
if root and str(path).startswith(root + os.sep):
return side + "/" + os.path.relpath(str(path), root).replace(os.sep, "/")
return None
def extract(work, project, package):
import griffe
from importlib import metadata
messages = []
class Collect(logging.Handler):
def emit(self, record):
messages.append(record)
log = logging.getLogger("griffe")
log.handlers[:] = [Collect()]
log.propagate = False
log.setLevel(logging.WARNING)
clones = {side: os.path.realpath(os.path.join(work, side)) for side in SIDES}
result = {"sides": {}, "griffe": "griffelib " + metadata.version("griffelib")}
trees, roots = {}, {}
for side in SIDES:
clone = clones[side]
version, note = read_version(clone, project)
info = {"version": version, "versionSource": note, "layout": None, "modules": 0}
result["sides"][side] = info
root = detect_root(clone, project, package)
if root is None:
info["error"] = "PACKAGE_NOT_FOUND"
continue
info["layout"] = os.path.relpath(root, clone).replace(os.sep, "/")
roots[side] = clone
broken = unparsable(root, package)
if broken:
info["error"] = "ANALYSIS_FAILED"
info["message"] = "cannot parse " + side + "/" + info["layout"] + "/" + broken
continue
try:
tree = griffe.load(
package,
search_paths=[root],
try_relative_path=False,
allow_inspection=False,
force_inspection=False,
find_stubs_package=False,
resolve_aliases=False,
resolve_external=False,
store_source=False,
)
except ModuleNotFoundError:
info["error"] = "PACKAGE_NOT_FOUND"
continue
except Exception as exc: # syntax errors and anything else the static loader cannot handle
info["error"] = "ANALYSIS_FAILED"
info["message"] = short(type(exc).__name__ + ": " + str(exc).replace(clone, side), 300)
continue
files = set()
for module in modules(tree):
paths = module.filepath if isinstance(module.filepath, list) else [module.filepath]
files.update(os.path.realpath(str(p)) for p in paths)
outside = sorted(f for f in files if not f.startswith(clone + os.sep))
if outside:
info["error"] = "UNSAFE_REPOSITORY"
info["message"] = str(len(outside)) + " loaded file(s) resolve outside the clone"
continue
info["modules"] = len(files)
trees[side] = tree
if len(trees) == 2:
old, new = trees["base"], trees["candidate"]
old_targets, new_targets = {}, {}
before, after = surface(old, package, {}, old_targets), surface(new, package, {}, new_targets)
breaking, informational, seen = [], [], set()
def record(b, path):
data = b.as_dict()
kind = b.kind.value
item = {"path": path, "kind": kind, "file": None, "line": None}
for key in ("old", "new"):
item[key] = short(_scrub(describe(data.get(key + "_value")) or "", clones)) or None
try:
item["file"] = relative(os.path.realpath(str(b.obj.filepath)), roots)
item["line"] = b.obj.lineno
except Exception:
pass
ident = (item["path"], kind, item["old"], item["new"])
if ident not in seen:
seen.add(ident)
(informational if kind in INFORMATIONAL else breaking).append(item)
for b in griffe.find_breaking_changes(old, new):
record(b, b.as_dict()["object_path"])
# Griffe skips objects that are only reachable through a re-export (for example a function defined in a
# private module and imported into __init__.py), so compare each re-exported object present on both sides.
cache = {}
def target_breakages(a, b):
# find_breaking_changes walks members, so compare the modules or classes that hold the two targets.
if a.parent is None or b.parent is None or a.name != b.name:
return [(x, a.path) for x in griffe.find_breaking_changes(a, b)]
key = (a.parent.path, b.parent.path)
if key not in cache:
cache[key] = list(griffe.find_breaking_changes(a.parent, b.parent))
found = []
for x in cache[key]:
object_path = x.as_dict()["object_path"]
for own in (a.path, b.path):
if object_path == own or object_path.startswith(own + "."):
found.append((x, own))
break
return found
for path in sorted(set(old_targets) & set(new_targets)):
for b, own in target_breakages(old_targets[path], new_targets[path]):
record(b, path + b.as_dict()["object_path"][len(own):])
flagged = {x["path"] for x in breaking + informational}
removed_paths = sorted(set(before) - set(after))
removed_set = set(removed_paths)
for path in removed_paths:
if path in flagged or any(path.startswith(p + ".") for p in removed_set if p != path):
continue
if any(path.startswith(f + ".") for f in flagged):
continue
entry = before[path]
breaking.append({
"path": path,
"kind": "Public re-export was removed" if entry["kind"] == "re-export" else "Public object was removed",
"old": entry.get("target"),
"new": None,
"file": None,
"line": None,
})
additions = []
for path in sorted(set(after) - set(before)):
parent = path.rsplit(".", 1)[0]
if parent in after and parent not in before:
continue
additions.append({"path": path, "kind": "Public " + after[path]["kind"] + " was added"})
for path in sorted(set(before) & set(after)):
if path in flagged or "parameters" not in before[path] or "parameters" not in after[path]:
continue
known = {p[0] for p in before[path]["parameters"]}
for name, kind, has_default in after[path]["parameters"]:
if name not in known and has_default:
additions.append({"path": path, "kind": "Optional parameter was added", "parameter": name})
key = lambda x: (x["path"], x["kind"], x.get("old") or "", x.get("new") or "", x.get("parameter") or "")
result["breaking"] = sorted(breaking, key=key)
result["additions"] = sorted(additions, key=key)
result["informational"] = sorted(informational, key=key)
result["surface"] = {"base": len(before), "candidate": len(after)}
result["griffeMessages"] = {
"count": len(messages),
"first": [short(r.levelname + ": " + _scrub(r.getMessage(), clones), 300) for r in messages[:5]],
}
return result
def _scrub(text, clones):
for side, clone in clones.items():
text = text.replace(clone, side)
return text
def bump(old, new):
from packaging.version import Version
a, b = Version(old), Version(new)
if b <= a:
return "none"
if b.major > a.major:
return "major"
if b.minor > a.minor:
return "minor"
return "patch"
def classify(env, api):
inconclusive, block, warn, notes = [], [], [], []
sides = (api or {}).get("sides", {})
verify_ok = env["VERIFY_SAFE"] == "true"
if not verify_ok:
inconclusive.append("UNSAFE_REPOSITORY")
elif api is None:
inconclusive.append("ANALYSIS_TIMEOUT" if env["EXTRACT_RC"] == "124" else "ANALYSIS_FAILED")
else:
for side in SIDES:
error = sides.get(side, {}).get("error")
if error and error not in inconclusive:
inconclusive.append(error)
versions = {}
for side, given_key in (("base", "BASE_VERSION"), ("candidate", "PROPOSED_VERSION")):
static = sides.get(side, {}).get("version")
source = sides.get(side, {}).get("versionSource")
given = env[given_key] or None
if given and static and given != static:
block.append("VERSION_MISMATCH")
notes.append(side + " pyproject.toml says " + static + " but " + given_key.lower() + " is " + given + ".")
versions[side] = {"value": given or static, "source": "input" if given else ("pyproject.toml" if static else None), "pyproject": static, "pyprojectNote": source}
old, new = versions["base"]["value"], versions["candidate"]["value"]
if verify_ok and (not old or not new):
inconclusive.append("VERSION_UNKNOWN")
breaking = (api or {}).get("breaking")
additions = (api or {}).get("additions")
analysed = breaking is not None
actual = required = None
if old and new:
actual = bump(old, new)
if actual == "none":
block.append("VERSION_NOT_INCREASING")
if analysed and old:
zero = old.split(".")[0] == "0"
if breaking:
required = "minor" if zero else "major"
elif additions and not zero:
required = "minor"
else:
required = "patch"
if actual and actual != "none":
if breaking and zero and actual == "patch":
block.append("BREAKING_CHANGE_REQUIRES_MINOR_0X")
elif breaking and not zero and actual != "major":
block.append("BREAKING_CHANGE_REQUIRES_MAJOR")
elif not breaking and additions and not zero and actual == "patch":
warn.append("NEW_API_REQUIRES_MINOR")
elif not breaking and not additions and not zero and actual == "major":
warn.append("UNNECESSARY_MAJOR")
block = list(dict.fromkeys(block))
verdict = "INCONCLUSIVE" if inconclusive else "BLOCK" if block else "WARN" if warn else "PASS"
reasons = {"INCONCLUSIVE": inconclusive, "BLOCK": block, "WARN": warn, "PASS": []}[verdict]
return verdict, reasons, versions, actual, required, notes
DESCRIBE = {
"BREAKING_CHANGE_REQUIRES_MAJOR": "The public API has breaking changes, so the major version must increase.",
"BREAKING_CHANGE_REQUIRES_MINOR_0X": "The public API has breaking changes. Under 0.x the minor version must increase at least.",
"NEW_API_REQUIRES_MINOR": "New public API was added in a patch release; a minor release is expected.",
"UNNECESSARY_MAJOR": "The major version increases but no public API change was detected.",
"VERSION_NOT_INCREASING": "The proposed version is not greater than the base version.",
"VERSION_MISMATCH": "An explicit version input disagrees with `[project].version` in `pyproject.toml`.",
"VERSION_UNKNOWN": "A version could not be read statically. Pass `base_version` or `proposed_version`, or use a static `[project].version`.",
"PACKAGE_NOT_FOUND": "The package was not found at `<project_path>/src/<package>` or `<project_path>/<package>` in one of the commits.",
"ANALYSIS_TIMEOUT": "Static analysis did not finish within the time limit.",
"ANALYSIS_FAILED": "The source could not be analysed statically (for example a syntax error or the memory limit).",
"UNSAFE_REPOSITORY": "The package contains symbolic links or `.pth` files, or a loaded file resolves outside the clone. Nothing was analysed.",
}
def render(report, execution):
rows = []
changes = report["changes"]
cap = 15
def listing(title, items, fmt):
if not items:
return []
lines = ["", "**" + title + "** (" + str(len(items)) + ")", ""]
lines += ["- " + fmt(x) for x in items[:cap]]
if len(items) > cap:
lines.append("- ... " + str(len(items) - cap) + " more in report.json")
return lines
def change(x):
where = " (" + x["file"] + ":" + str(x["line"]) + ")" if x.get("file") and x.get("line") else ""
code = lambda v: "`" + short(v, 60).replace("`", "'") + "`"
old, new = x.get("old"), x.get("new")
detail = ""
if x.get("parameter"):
detail = ": " + code(x["parameter"])
elif old and new:
detail = ": " + code(old) + " -> " + code(new)
elif old or new:
detail = ": " + code(old or new)
return "`" + x["path"] + "`: " + x["kind"] + detail + where
v = report["versions"]
rows += [
"### Python Release SemVer Gate: " + report["verdict"],
"",
"| | |",
"|---|---|",
"| Package | `" + report["package"] + "` in " + report["repository"]["slug"] + " |",
"| Commits | `" + report["repository"]["baseCommit"][:12] + "` -> `" + report["repository"]["candidateCommit"][:12] + "` |",
"| Version | " + (v["base"] or "?") + " -> " + (v["proposed"] or "?") + " (" + (report["policy"]["actualBump"] or "unknown") + " bump) |",
"| Required bump | " + (report["policy"]["requiredBump"] + " (" + report["policy"]["series"] + " policy)" if report["policy"]["requiredBump"] else "unknown") + " |",
"| API changes | " + str(len(changes["breaking"])) + " breaking, " + str(len(changes["additions"])) + " added, " + str(len(changes["informational"])) + " informational |",
]
if report["reasonCodes"]:
rows += ["", "**Reasons**", ""] + ["- `" + c + "`: " + DESCRIBE[c] for c in report["reasonCodes"]]
rows += ["- " + n for n in report["notes"]] if report["notes"] else []
rows += listing("Breaking changes", changes["breaking"], change)
rows += listing("Additions", changes["additions"], change)
rows += [
"",
"<sub>Kestra execution " + execution + ". Static analysis of the public API with " + report["analysis"]["tool"]
+ "; repository code is never imported, installed or run. Strict SemVer policy: some projects deliberately remove deprecated APIs in minor releases.</sub>",
]
return "\n".join(rows)
def main():
mode = sys.argv[1]
env = os.environ
if mode == "extract":
import resource
import time
started = time.monotonic()
try:
api = extract(env["WORK"], env["PROJECT_PATH"], env["PACKAGE"])
except Exception as exc:
api = {"sides": {s: {"error": "ANALYSIS_FAILED"} for s in SIDES}, "message": short(type(exc).__name__, 100)}
# Defence in depth: no absolute path of this run's workspace may reach the report.
text = json.dumps(api, sort_keys=True).replace(os.path.realpath(env["WORK"]) + "/", "")
with open(env["OUT"] + "/api.json", "w") as f:
f.write(text)
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss // 1024
print("Static analysis finished in %.1fs, peak memory %d MB" % (time.monotonic() - started, peak))
return
api_path = env["OUT"] + "/api.json"
api = None
if env["VERIFY_SAFE"] == "true" and env["EXTRACT_RC"] == "0" and os.path.isfile(api_path):
with open(api_path) as f:
api = json.load(f)
verdict, reasons, versions, actual, required, notes = classify(env, api)
sides = (api or {}).get("sides", {})
for side in SIDES:
if sides.get(side, {}).get("message"):
notes.append(side + ": " + sides[side]["message"])
old = versions["base"]["value"]
slug = re.sub(r"\.git$", "", env["REPO_URL"].replace("https://github.com/", ""))
report = {
"schema": SCHEMA,
"verdict": verdict,
"reasonCodes": reasons,
"notes": notes,
"repository": {"url": env["REPO_URL"], "slug": slug, "baseCommit": env["BASE_COMMIT"], "candidateCommit": env["CANDIDATE_COMMIT"]},
"package": env["PACKAGE"],
"projectPath": env["PROJECT_PATH"],
"versions": {
"base": old,
"baseSource": versions["base"]["source"],
"proposed": versions["candidate"]["value"],
"proposedSource": versions["candidate"]["source"],
"pyproject": {s: {"version": versions[k]["pyproject"], "note": versions[k]["pyprojectNote"]} for s, k in (("base", "base"), ("candidate", "candidate"))},
},
"policy": {
"name": "strict-semver",
"series": ("0.x" if old.split(".")[0] == "0" else ">=1.0") if old else "unknown",
"requiredBump": required,
"actualBump": actual,
},
"changes": {
"breaking": (api or {}).get("breaking") or [],
"additions": (api or {}).get("additions") or [],
"informational": (api or {}).get("informational") or [],
},
"analysis": {
"completed": api is not None and "breaking" in api,
"tool": (api or {}).get("griffe") or "griffelib 2.3.0",
"python": "%d.%d.%d" % sys.version_info[:3],
"allowInspection": False,
"network": "none",
"user": "nobody",
"layout": {s: sides.get(s, {}).get("layout") for s in SIDES},
"modules": {s: sides.get(s, {}).get("modules", 0) for s in SIDES},
"publicSurface": (api or {}).get("surface"),
"griffeMessages": (api or {}).get("griffeMessages", {"count": 0, "first": []}),
},
"security": {
"verified": env["VERIFY_SAFE"] == "true",
"problems": [p for p in env.get("VERIFY_PROBLEMS", "").split("|") if p],
"symlinksRemovedOutsidePackage": int(env.get("VERIFY_LINKS_REMOVED") or 0),
},
}
with open("report.json", "w") as f:
json.dump(report, f, indent=2, sort_keys=True)
f.write("\n")
body = render(report, env["EXECUTION_ID"])
counts = {k: len(v) for k, v in report["changes"].items()}
print("Verdict " + verdict + (": " + ", ".join(reasons) if reasons else "") + " | " + json.dumps(counts, sort_keys=True))
print("::" + json.dumps({"outputs": {"verdict": verdict, "reasonCodes": reasons, "repository": slug, "requiredBump": required, "actualBump": actual, "counts": counts, "commentBody": body}}) + "::")
if __name__ == "__main__":
main()
tasks:
- id: fetch_tools
type: io.kestra.plugin.scripts.shell.Commands
description: Downloads the two pinned, hash-checked wheels from PyPI. Runs
before anything is cloned, so the only container with network access
never sees repository content.
timeout: PT5M
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python@sha256:05cda9777409a9c3ffddd94a4c476b79f0769a0b4857f0c7ed9226b6800b0d6f
commands:
- |
pip download --no-deps --only-binary=:all: --require-hashes --no-cache-dir --disable-pip-version-check --index-url https://pypi.org/simple -r requirements.txt -d wheels
- id: clone_base
type: io.kestra.plugin.git.Clone
description: Checks out the exact base commit, without submodules.
timeout: PT10M
url: "{{ inputs.repo_url }}"
commit: "{{ inputs.base_commit }}"
cloneSubmodules: false
directory: base
- id: clone_candidate
type: io.kestra.plugin.git.Clone
description: Checks out the exact candidate commit, without submodules.
timeout: PT10M
url: "{{ inputs.repo_url }}"
commit: "{{ inputs.candidate_commit }}"
cloneSubmodules: false
directory: candidate
- id: verify
type: io.kestra.plugin.scripts.shell.Commands
description: Runs on the Kestra worker before the analysis container starts.
Checks that each checkout is the requested commit, refuses symbolic
links and .pth files inside the analysed package, deletes symbolic
links everywhere else so no container can follow them, keeps
project_path inside each clone, and removes the .git folders. A
refused repository is deleted and reported as INCONCLUSIVE.
taskRunner:
type: io.kestra.plugin.core.runner.Process
env:
BASE_COMMIT: "{{ inputs.base_commit }}"
CANDIDATE_COMMIT: "{{ inputs.candidate_commit }}"
PROJECT_PATH: "{{ inputs.project_path }}"
PACKAGE: "{{ inputs.package }}"
commands:
- |
set -eu
# This task runs on the Kestra worker, whose environment can hold secrets. Keep only what the checks need.
for NAME in $(export -p | sed -n -e 's/^export \([A-Za-z_][A-Za-z0-9_]*\)=.*/\1/p' -e 's/^declare -x \([A-Za-z_][A-Za-z0-9_]*\)=.*/\1/p'); do
case "$NAME" in BASE_COMMIT|CANDIDATE_COMMIT|PROJECT_PATH|PACKAGE|PATH|HOME|PWD) ;; *) unset "$NAME" 2>/dev/null || true ;; esac
done
PROBLEMS=""; REMOVED=0
for side in base candidate; do
if [ "$side" = base ]; then WANT=$BASE_COMMIT; else WANT=$CANDIDATE_COMMIT; fi
HEAD_SHA=$(cat "$side/.git/HEAD")
if [ "$HEAD_SHA" != "$WANT" ]; then echo "$side checkout is at '$HEAD_SHA', not the requested commit $WANT."; exit 1; fi
ROOT=$(realpath -e "$side"); PROJECT=$(realpath -m -- "$side/$PROJECT_PATH")
case "$PROJECT/" in "$ROOT"/*) ;; *) PROBLEMS="$PROBLEMS|$side: project_path resolves outside the clone"; continue ;; esac
rm -rf "$side/.git"
# The only files the analysis reads: pyproject.toml and the package in <project>/src or <project>.
SCOPE=0; PTH=0
for d in "$PROJECT/src" "$PROJECT"; do
for p in "$d/$PACKAGE" "$d/$PACKAGE.py"; do
if [ -L "$p" ] || [ -L "$d" ]; then SCOPE=$((SCOPE + 1)); elif [ -d "$p" ]; then SCOPE=$((SCOPE + $(find "$p" -type l | wc -l))); PTH=$((PTH + $(find "$p" -name '*.pth' | wc -l))); fi
done
if [ -d "$d" ] && [ ! -L "$d" ]; then PTH=$((PTH + $(find "$d" -maxdepth 1 -name '*.pth' | wc -l))); fi
done
if [ -L "$PROJECT/pyproject.toml" ]; then SCOPE=$((SCOPE + 1)); fi
if [ "$SCOPE" != 0 ]; then PROBLEMS="$PROBLEMS|$side: $SCOPE symbolic link(s) in the package or pyproject.toml"; fi
if [ "$PTH" != 0 ]; then PROBLEMS="$PROBLEMS|$side: $PTH .pth file(s) where the package is looked up"; fi
# Links elsewhere (docs, tests) are deleted so that no later task can follow them out of the clone.
N=$(find "$side" -type l | wc -l); REMOVED=$((REMOVED + N)); find "$side" -type l -delete
done
SAFE=true
if [ -n "$PROBLEMS" ]; then SAFE=false; rm -rf base candidate; echo "Refusing to analyse: ${PROBLEMS#|}"; else echo "Verified both checkouts: exact commits, no symbolic link or .pth file in the package, project_path inside the clone. Symbolic links removed elsewhere: $REMOVED."; fi
printf '::{"outputs":{"safe":%s,"problems":"%s","linksRemoved":%s}}::\n' "$SAFE" "${PROBLEMS#|}" "$REMOVED"
- id: analyze
type: io.kestra.plugin.scripts.shell.Commands
description: Offline and without secrets. Installs the wheels with their hashes
checked, loads both commits with Griffe in static mode as the
unprivileged nobody user, compares the public APIs, applies the SemVer
policy and writes report.json.
timeout: PT10M
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: none
cpu:
cpus: 2
memory:
memory: 1GB
containerImage: python@sha256:05cda9777409a9c3ffddd94a4c476b79f0769a0b4857f0c7ed9226b6800b0d6f
env:
VERIFY_SAFE: "{{ outputs.verify.vars.safe }}"
VERIFY_PROBLEMS: "{{ outputs.verify.vars.problems ?? '' }}"
VERIFY_LINKS_REMOVED: "{{ outputs.verify.vars.linksRemoved ?? 0 }}"
REPO_URL: "{{ inputs.repo_url }}"
BASE_COMMIT: "{{ inputs.base_commit }}"
CANDIDATE_COMMIT: "{{ inputs.candidate_commit }}"
PACKAGE: "{{ inputs.package }}"
PROJECT_PATH: "{{ inputs.project_path }}"
BASE_VERSION: "{{ inputs.base_version ?? '' }}"
PROPOSED_VERSION: "{{ inputs.proposed_version ?? '' }}"
EXECUTION_ID: "{{ execution.id }}"
outputFiles:
- report.json
commands:
- |
set -u
W=$PWD; OUT=/tmp/gate-out
mkdir -p "$OUT" /opt/gate && chown 65534:65534 "$OUT"
pip install --no-index --find-links wheels --require-hashes --no-deps --target /opt/gate -r requirements.txt --quiet --disable-pip-version-check --root-user-action=ignore
EXTRACT_RC=-1; LIMIT=480
if [ "$VERIFY_SAFE" = true ]; then
EXTRACT_RC=0
timeout -k 10 "$LIMIT" setpriv --reuid=65534 --regid=65534 --clear-groups env -i PATH="$PATH" HOME=/tmp WORK="$W" OUT="$OUT" PROJECT_PATH="$PROJECT_PATH" PACKAGE="$PACKAGE" python -I -B gate.py extract || EXTRACT_RC=$?
fi
export EXTRACT_RC OUT
python -I -B gate.py classify
- id: comment_on_pr
type: io.kestra.plugin.core.flow.If
description: Only touches GitHub when a pull request number was given.
condition: "{{ (inputs.pr_number ?? 0) > 0 }}"
then:
- id: pr_comment
type: io.kestra.plugin.github.issues.Comment
description: One comment per pull request, project and package. Later runs
update it in place instead of adding a new one.
oauthToken: "{{ secret('GITHUB_TOKEN') }}"
repository: "{{ outputs.analyze.vars.repository }}"
issueNumber: "{{ inputs.pr_number }}"
updateTag: "python-release-semver-gate:{{ inputs.project_path }}:{{
inputs.package }}"
body: "{{ outputs.analyze.vars.commentBody }}"
- id: verdict
type: io.kestra.plugin.core.flow.Switch
description: BLOCK fails the execution so a release pipeline waiting on it
stops; INCONCLUSIVE ends in WARNING; PASS and WARN succeed.
value: "{{ outputs.analyze.vars.verdict }}"
cases:
BLOCK:
- id: block
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Release of {{ inputs.package }} is blocked: {{
outputs.analyze.vars.reasonCodes | join(', ') }}"
INCONCLUSIVE:
- id: inconclusive
type: io.kestra.plugin.core.flow.AllowFailure
description: Ends the execution in WARNING rather than FAILED, and still
publishes the flow outputs.
tasks:
- id: not_determined
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Release of {{ inputs.package }} could not be checked: {{
outputs.analyze.vars.reasonCodes | join(', ') }}"
defaults:
- id: release_ok
type: io.kestra.plugin.core.log.Log
message: "Release of {{ inputs.package }} ({{ outputs.analyze.vars.actualBump }}
bump): {{ outputs.analyze.vars.verdict }}{{
(outputs.analyze.vars.reasonCodes | length) > 0 ? ' (' ~
(outputs.analyze.vars.reasonCodes | join(', ')) ~ ')' : '' }}"
outputs:
- id: verdict
type: STRING
description: PASS, WARN, BLOCK or INCONCLUSIVE.
value: "{{ outputs.analyze.vars.verdict ?? 'ERROR' }}"
- id: reason_codes
type: JSON
description: Reason codes behind the verdict, for example
["BREAKING_CHANGE_REQUIRES_MAJOR"].
value: "{{ (outputs.analyze.vars.reasonCodes ?? []) | toJson }}"
- id: required_bump
type: STRING
description: Smallest version bump the detected API changes require (major,
minor or patch).
value: "{{ outputs.analyze.vars.requiredBump ?? 'unknown' }}"
- id: report
type: FILE
description: report.json, schema kestra-python-release-semver-gate/1.0.
value: "{{ outputs.analyze.outputFiles['report.json'] ?? '' }}"
triggers:
- id: release_webhook
type: io.kestra.plugin.core.trigger.Webhook
description: 'POST {"repo_url", "base_commit", "candidate_commit", "package",
"project_path", "base_version", "proposed_version", "pr_number"} from CI,
for example when a release pull request is opened. Values go through the
same validators as a manual run; an invalid request gets HTTP 422 and no
execution.'
key: "{{ secret('SEMVER_GATE_WEBHOOK_KEY') }}"
inputs:
repo_url: "{{ trigger.body.repo_url }}"
base_commit: "{{ trigger.body.base_commit }}"
candidate_commit: "{{ trigger.body.candidate_commit }}"
package: "{{ trigger.body.package }}"
project_path: "{{ trigger.body.project_path ?? '.' }}"
base_version: "{{ trigger.body.base_version ?? '' }}"
proposed_version: "{{ trigger.body.proposed_version ?? '' }}"
pr_number: "{{ trigger.body.pr_number ?? '' }}"
