fix: fetch rosdistro yaml from <distro>/distro.yaml
This commit is contained in:
@@ -17,9 +17,6 @@ def _raw_url(base_url: str, owner: str, repo: str, branch: str, file_path: str)
|
|||||||
|
|
||||||
|
|
||||||
def _parse_package_ref(value: str) -> tuple[str, str, str]:
|
def _parse_package_ref(value: str) -> tuple[str, str, str]:
|
||||||
"""
|
|
||||||
Parses "owner/repo@ref" into (owner, repo, ref).
|
|
||||||
"""
|
|
||||||
value = value.strip()
|
value = value.strip()
|
||||||
|
|
||||||
if "@" not in value:
|
if "@" not in value:
|
||||||
@@ -45,6 +42,15 @@ def _parse_package_ref(value: str) -> tuple[str, str, str]:
|
|||||||
return owner, repo, ref
|
return owner, repo, ref
|
||||||
|
|
||||||
|
|
||||||
|
def _fetch_text(url: str, timeout_s: float) -> str:
|
||||||
|
resp = requests.get(url, timeout=timeout_s)
|
||||||
|
if resp.status_code != 200:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Failed to fetch '{url}' (HTTP {resp.status_code}): {resp.text}"
|
||||||
|
)
|
||||||
|
return resp.text
|
||||||
|
|
||||||
|
|
||||||
def fetch_rosdistro_packages(
|
def fetch_rosdistro_packages(
|
||||||
*,
|
*,
|
||||||
base_url: str,
|
base_url: str,
|
||||||
@@ -55,22 +61,42 @@ def fetch_rosdistro_packages(
|
|||||||
timeout_s: float = 20.0,
|
timeout_s: float = 20.0,
|
||||||
) -> Dict[str, Package]:
|
) -> Dict[str, Package]:
|
||||||
"""
|
"""
|
||||||
Fetches rosdistro/<rosdistro>/distro.yaml from a Gitea repo and parses it into:
|
Tries these paths (in order):
|
||||||
|
1) <rosdistro>/distro.yaml (your current layout)
|
||||||
|
2) rosdistro/<rosdistro>/distro.yaml (fallback / older layout)
|
||||||
|
|
||||||
|
Returns:
|
||||||
{ package_name: Package(name, owner, repo, ref) }
|
{ package_name: Package(name, owner, repo, ref) }
|
||||||
"""
|
"""
|
||||||
file_path = f"rosdistro/{rosdistro}/distro.yaml"
|
candidates = [
|
||||||
url = _raw_url(base_url, owner, repo, branch, file_path)
|
f"{rosdistro}/distro.yaml",
|
||||||
|
f"rosdistro/{rosdistro}/distro.yaml",
|
||||||
|
]
|
||||||
|
|
||||||
resp = requests.get(url, timeout=timeout_s)
|
last_err: Exception | None = None
|
||||||
if resp.status_code != 200:
|
text: str | None = None
|
||||||
raise RuntimeError(
|
used_path: str | None = None
|
||||||
f"Failed to fetch '{url}' (HTTP {resp.status_code}): {resp.text}"
|
|
||||||
|
for file_path in candidates:
|
||||||
|
url = _raw_url(base_url, owner, repo, branch, file_path)
|
||||||
|
try:
|
||||||
|
text = _fetch_text(url, timeout_s)
|
||||||
|
used_path = file_path
|
||||||
|
break
|
||||||
|
except Exception as e:
|
||||||
|
last_err = e
|
||||||
|
|
||||||
|
if text is None or used_path is None:
|
||||||
|
raise (
|
||||||
|
last_err
|
||||||
|
if last_err is not None
|
||||||
|
else RuntimeError("Failed to fetch rosdistro YAML")
|
||||||
)
|
)
|
||||||
|
|
||||||
data = yaml.safe_load(resp.text)
|
data = yaml.safe_load(text)
|
||||||
if not isinstance(data, dict):
|
if not isinstance(data, dict):
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
f"Invalid YAML structure in {file_path}: expected mapping at root"
|
f"Invalid YAML structure in {used_path}: expected mapping at root"
|
||||||
)
|
)
|
||||||
|
|
||||||
packages: Dict[str, Package] = {}
|
packages: Dict[str, Package] = {}
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from typing import Any, Dict
|
from typing import Dict
|
||||||
|
|
||||||
from ros2cli.verb import VerbExtension
|
from ros2cli.verb import VerbExtension
|
||||||
|
|
||||||
@@ -8,7 +8,7 @@ from ..package import Package
|
|||||||
|
|
||||||
|
|
||||||
class ListVerb(VerbExtension):
|
class ListVerb(VerbExtension):
|
||||||
"""List packages from rosdistro/<rosdistro>/distro.yaml."""
|
"""List packages from <rosdistro>/distro.yaml."""
|
||||||
|
|
||||||
def add_arguments(self, parser, cli_name):
|
def add_arguments(self, parser, cli_name):
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
@@ -22,9 +22,17 @@ class ListVerb(VerbExtension):
|
|||||||
packages: Dict[str, Package] = getattr(args, "packages", {})
|
packages: Dict[str, Package] = getattr(args, "packages", {})
|
||||||
|
|
||||||
if args.format == "yaml":
|
if args.format == "yaml":
|
||||||
for name in sorted(packages.keys()):
|
out = {}
|
||||||
pkg = packages[name]
|
for name, pkg in packages.items():
|
||||||
print(f"{name}: {pkg.owner}/{pkg.repo}@{pkg.ref}")
|
out[name] = {
|
||||||
|
"owner": pkg.owner,
|
||||||
|
"repo": pkg.repo,
|
||||||
|
"ref": pkg.ref,
|
||||||
|
}
|
||||||
|
|
||||||
|
import yaml
|
||||||
|
|
||||||
|
print(yaml.dump(out, sort_keys=True))
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
for name in sorted(packages.keys()):
|
for name in sorted(packages.keys()):
|
||||||
@@ -36,6 +44,11 @@ class ListVerb(VerbExtension):
|
|||||||
|
|
||||||
def add_list_verb(subparsers):
|
def add_list_verb(subparsers):
|
||||||
parser = subparsers.add_parser("list", help="List packages from rosdistro")
|
parser = subparsers.add_parser("list", help="List packages from rosdistro")
|
||||||
|
|
||||||
verb = ListVerb()
|
verb = ListVerb()
|
||||||
verb.add_arguments(parser, "buildfarm")
|
verb.add_arguments(parser, "buildfarm")
|
||||||
parser.set_defaults(main=verb.main)
|
|
||||||
|
def _run_list_verb(args):
|
||||||
|
return verb.main(args=args)
|
||||||
|
|
||||||
|
parser.set_defaults(main=_run_list_verb)
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
<?xml version="1.0"?>
|
<?xml version="1.0"?>
|
||||||
<package format="3">
|
<package format="3">
|
||||||
<name>buildfarm</name>
|
<name>buildfarm</name>
|
||||||
<version>0.0.1</version>
|
<version>0.0.2</version>
|
||||||
<description>Scripts to build ROS2 packages as .deb packages.</description>
|
<description>Scripts to build ROS2 packages as .deb packages.</description>
|
||||||
<maintainer email="mail@maikknof.de">Maik Knof</maintainer>
|
<maintainer email="mail@maikknof.de">Maik Knof</maintainer>
|
||||||
<license>TODO</license>
|
<license>TODO</license>
|
||||||
|
|||||||
Reference in New Issue
Block a user