You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2025/07/08 16:19:24 UTC
(tooling-trusted-release) branch main updated: Adapt PubSub and SVN code from the BAT
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-release.git
The following commit(s) were added to refs/heads/main by this push:
new e73f2e1 Adapt PubSub and SVN code from the BAT
e73f2e1 is described below
commit e73f2e194052bd233b21c37f0e131dceab249f30
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Tue Jul 8 17:18:46 2025 +0100
Adapt PubSub and SVN code from the BAT
---
atr/config.py | 8 +++
atr/server.py | 33 ++++++++++
atr/svn/__init__.py | 180 ++++++++++++++++++++++++++++++++++++++++++++++++++++
atr/svn/pubsub.py | 97 ++++++++++++++++++++++++++++
poetry.lock | 22 ++++++-
pyproject.toml | 5 +-
6 files changed, 342 insertions(+), 3 deletions(-)
diff --git a/atr/config.py b/atr/config.py
index 2c6d6e8..3acc85a 100644
--- a/atr/config.py
+++ b/atr/config.py
@@ -44,6 +44,10 @@ class AppConfig:
STATE_DIR = decouple.config("STATE_DIR", default=os.path.join(PROJECT_ROOT, "state"))
LDAP_BIND_DN = _config_secrets("LDAP_BIND_DN", STATE_DIR, default=None, cast=str)
LDAP_BIND_PASSWORD = _config_secrets("LDAP_BIND_PASSWORD", STATE_DIR, default=None, cast=str)
+ PUBSUB_URL = _config_secrets("PUBSUB_URL", STATE_DIR, default=None, cast=str)
+ PUBSUB_USER = _config_secrets("PUBSUB_USER", STATE_DIR, default=None, cast=str)
+ PUBSUB_PASSWORD = _config_secrets("PUBSUB_PASSWORD", STATE_DIR, default=None, cast=str)
+ SVN_TOKEN = _config_secrets("SVN_TOKEN", STATE_DIR, default=None, cast=str)
DEBUG = False
TEMPLATES_AUTO_RELOAD = False
@@ -53,6 +57,9 @@ class AppConfig:
DOWNLOADS_STORAGE_DIR = os.path.join(STATE_DIR, "downloads")
FINISHED_STORAGE_DIR = os.path.join(STATE_DIR, "finished")
UNFINISHED_STORAGE_DIR = os.path.join(STATE_DIR, "unfinished")
+ # TODO: By convention this is at /x1/, but we can symlink it here perhaps?
+ # TODO: We need to get Puppet to check SVN out initially, or do it manually
+ SVN_STORAGE_DIR = os.path.join(STATE_DIR, "svn")
SQLITE_DB_PATH = decouple.config("SQLITE_DB_PATH", default="atr.db")
# Apache RAT configuration
@@ -123,6 +130,7 @@ def get() -> type[AppConfig]:
(config.DOWNLOADS_STORAGE_DIR, "DOWNLOADS_STORAGE_DIR"),
(config.FINISHED_STORAGE_DIR, "FINISHED_STORAGE_DIR"),
(config.UNFINISHED_STORAGE_DIR, "UNFINISHED_STORAGE_DIR"),
+ (config.SVN_STORAGE_DIR, "SVN_STORAGE_DIR"),
]
relative_paths = [
(config.SQLITE_DB_PATH, "SQLITE_DB_PATH"),
diff --git a/atr/server.py b/atr/server.py
index 62abdb9..167db3d 100644
--- a/atr/server.py
+++ b/atr/server.py
@@ -17,6 +17,8 @@
"""server.py"""
+import asyncio
+import contextlib
import logging
import os
from collections.abc import Iterable
@@ -43,6 +45,7 @@ import atr.filters as filters
import atr.manager as manager
import atr.preload as preload
import atr.ssh as ssh
+import atr.svn.pubsub as pubsub
import atr.template as template
import atr.user as user
import atr.util as util
@@ -147,6 +150,31 @@ def app_setup_lifecycle(app: base.QuartApp) -> None:
worker_manager = manager.get_worker_manager()
await worker_manager.start()
+ conf = config.get()
+ pubsub_url = conf.PUBSUB_URL
+ pubsub_user = conf.PUBSUB_USER
+ pubsub_password = conf.PUBSUB_PASSWORD
+
+ if pubsub_url and pubsub_user and pubsub_password:
+ app.logger.info("Starting SVN listener")
+ listener = pubsub.SVNListener(
+ working_copy_root=conf.SVN_STORAGE_DIR,
+ url=pubsub_url,
+ username=pubsub_user,
+ password=pubsub_password,
+ )
+ task = asyncio.create_task(listener.start())
+ app.extensions["svn_listener"] = task
+ app.logger.info("SVN listener task created")
+ else:
+ app.logger.info(
+ "SVN listener not started: url=%s user=%s password=%s",
+ bool(pubsub_url),
+ bool(pubsub_user),
+ # Essential to use bool(...) here to avoid logging the password
+ bool(pubsub_password),
+ )
+
ssh_server = await ssh.server_start()
app.extensions["ssh_server"] = ssh_server
@@ -160,6 +188,11 @@ def app_setup_lifecycle(app: base.QuartApp) -> None:
if ssh_server:
await ssh.server_stop(ssh_server)
+ if task := app.extensions.get("svn_listener"):
+ task.cancel()
+ with contextlib.suppress(asyncio.CancelledError):
+ await task
+
await db.shutdown_database()
app.background_tasks.clear()
diff --git a/atr/svn/__init__.py b/atr/svn/__init__.py
new file mode 100644
index 0000000..43f9625
--- /dev/null
+++ b/atr/svn/__init__.py
@@ -0,0 +1,180 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+import datetime
+import logging
+import pathlib
+from typing import Final, Self
+
+import pydantic
+import pydantic_xml
+
+import atr.config as config
+
+_ASF_TOOL: Final[str] = "atr"
+_LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
+
+
+class CommandExecutionError(RuntimeError):
+ # TODO: These are never assigned
+ returncode: int
+ output: str
+
+
+class SvnInfo(pydantic.BaseModel):
+ """A dataclass to hold information about a file in a subversion repository."""
+
+ path: str
+ name: str
+ url: str
+ relative_url: str
+ repository_root: str
+ revision: str
+ last_changed_author: str
+ last_changed_rev: str
+ last_changed_date: str
+ checksum: str | None = None
+ text_last_updated: str | None = None
+
+ @property
+ def revision_number(self) -> int:
+ return int(self.revision)
+
+ @property
+ def last_changed_rev_number(self) -> int:
+ return int(self.last_changed_rev)
+
+ @classmethod
+ async def from_url(cls, url: str) -> Self:
+ output = await _run_svn_info(url)
+
+ nfo = {}
+ for line in output.split("\n"):
+ # TODO: Might break on IPv6 hosts, or hosts with a port?
+ k, v = line.split(":", 1)
+ nfo[k.replace(" ", "_").lower()] = v.strip()
+
+ return cls.model_validate(nfo)
+
+ @classmethod
+ async def from_path(cls, path: pathlib.Path) -> Self:
+ return await cls.from_url(str(path))
+
+
+class SvnLogEntry(pydantic_xml.BaseXmlModel):
+ revision: int = pydantic_xml.attr()
+ author: str = pydantic_xml.element()
+ date: str = pydantic_xml.element()
+ msg: str | None = pydantic_xml.element(default=None)
+
+ @property
+ def datetime(self) -> datetime.datetime:
+ return datetime.datetime.fromisoformat(self.date)
+
+
+class SvnLog(pydantic_xml.BaseXmlModel, tag="log"):
+ entries: list[SvnLogEntry] = pydantic_xml.element(tag="logentry")
+
+
+async def _run_svnmucc_command(*args: str) -> str:
+ return await run_command("svnmucc", *args)
+
+
+async def _run_svn_command(sub_cmd: str, path: str, *args: str) -> str:
+ # Do not log this command, as it may contain a password or secret token
+ return await run_command("svn", *[sub_cmd, *args, path])
+
+
+async def _run_svn_info(path_or_url: str) -> str:
+ _LOGGER.debug(f"fetching svn info for '{path_or_url}'")
+ return await _run_svn_command("info", path_or_url)
+
+
+async def update(path: pathlib.Path) -> str:
+ _LOGGER.debug(f"running svn update for '{path}'")
+ return await _run_svn_command("update", str(path), "--parents")
+
+
+async def get_log(path: pathlib.Path) -> SvnLog:
+ _LOGGER.debug(f"running svn log for '{path}'")
+ svn_token = config.get().SVN_TOKEN
+ if svn_token is None:
+ raise ValueError("SVN_TOKEN must be set")
+ # TODO: Or omit username entirely?
+ log_output = await _run_svn_command("log", str(path), "--xml", "--username", _ASF_TOOL, "--password", svn_token)
+ return SvnLog.from_xml(log_output)
+
+
+async def get_diff(path: pathlib.Path, revision: int) -> str:
+ _LOGGER.debug(f"running svn diff for '{path}': r{revision}")
+ svn_token = config.get().SVN_TOKEN
+ if svn_token is None:
+ raise ValueError("SVN_TOKEN must be set")
+ # TODO: Or omit username entirely?
+ return await _run_svn_command(
+ "diff", str(path), "-c", str(revision), "--username", _ASF_TOOL, "--password", svn_token
+ )
+
+
+async def commit(path: pathlib.Path, url: str, username: str, revision: str, message: str) -> str:
+ _LOGGER.debug(f"running svn commit for user '{username}' to '{url}'")
+ # The username here is the ASF UID of the committer
+ svn_token = config.get().SVN_TOKEN
+ if svn_token is None:
+ raise ValueError("SVN_TOKEN must be set")
+ return await _run_svnmucc_command(
+ "put",
+ str(path),
+ url,
+ "--username",
+ username,
+ "--password",
+ svn_token,
+ "--non-interactive",
+ "--with-revprop",
+ f"asf:tool={_ASF_TOOL}",
+ "-r",
+ revision,
+ "-m",
+ message,
+ )
+
+
+async def run_command(cmd: str, *args: str) -> str:
+ """Run a svn command asynchronously.
+
+ Arguments:
+ cmd (str): the command to run
+ *args (str): arguments to pass to the command
+ """
+ proc = await asyncio.create_subprocess_exec(
+ cmd,
+ *args,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ )
+
+ stdout, stderr = await proc.communicate()
+
+ # if the proc.communicate() call returns an error
+ # print the error out and return an empty string.
+ if proc.returncode:
+ raise CommandExecutionError(proc.returncode, stderr.decode())
+ else:
+ output = stdout.decode().strip()
+ return output
diff --git a/atr/svn/pubsub.py b/atr/svn/pubsub.py
new file mode 100644
index 0000000..a75d95e
--- /dev/null
+++ b/atr/svn/pubsub.py
@@ -0,0 +1,97 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import logging
+import os
+import pathlib
+import urllib.parse
+from typing import TYPE_CHECKING, Final
+
+import asfpy.pubsub
+
+import atr.svn as svn
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+_LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
+# TODO: Check that these prefixes are correct
+_WATCHED_PREFIXES: Final[tuple[str, ...]] = (
+ "/svn/dist/dev",
+ "/svn/dist/release",
+)
+
+
+class SVNListener:
+ def __init__(
+ self,
+ working_copy_root: os.PathLike | str,
+ url: str,
+ username: str,
+ password: str,
+ topics: str = "commit/svn",
+ ) -> None:
+ self.working_copy_root = pathlib.Path(working_copy_root)
+ self.url = url
+ self.username = username
+ self.password = password
+ self.topics = topics
+
+ async def start(self) -> None:
+ """Run forever, processing PubSub payloads as they arrive."""
+ # TODO: Add reconnection logic here?
+ # Or does asfpy.pubsub.listen() already do this?
+ async for payload in asfpy.pubsub.listen(
+ # TODO: Upstream this change to BAT
+ urllib.parse.urljoin(self.url, self.topics),
+ username=self.username,
+ password=self.password,
+ ):
+ if (payload is None) or ("stillalive" in payload):
+ continue
+
+ pubsub_path = str(payload.get("pubsub_path", ""))
+ if not pubsub_path.startswith(_WATCHED_PREFIXES):
+ # Ignore commits outside dist/dev or dist/release
+ continue
+
+ _LOGGER.debug("PubSub payload: %s", payload)
+ await self._process_payload(payload)
+
+ async def _process_payload(self, payload: dict) -> None:
+ """
+ Update each changed file in the local working copy.
+
+ Payload format that we listen to:
+ {
+ "commit": {
+ "changed": ["/path/inside/repo/foo.txt", ...]
+ },
+ ...
+ }
+ """
+ changed: Sequence[str] = payload.get("commit", {}).get("changed", [])
+ for repo_path in changed:
+ prefix = next((p for p in _WATCHED_PREFIXES if repo_path.startswith(p)), "")
+ if not prefix:
+ continue
+ local_path = self.working_copy_root / repo_path[len(prefix) :].lstrip("/")
+ try:
+ await svn.update(local_path)
+ _LOGGER.info("svn updated %s", local_path)
+ except Exception as exc:
+ _LOGGER.warning("failed svn update %s: %s", local_path, exc)
diff --git a/poetry.lock b/poetry.lock
index 96a0ccd..a7f532b 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -2131,6 +2131,26 @@ files = [
[package.dependencies]
typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0"
+[[package]]
+name = "pydantic-xml"
+version = "2.17.2"
+description = "pydantic xml extension"
+optional = false
+python-versions = ">=3.8"
+groups = ["main"]
+files = [
+ {file = "pydantic_xml-2.17.2-py3-none-any.whl", hash = "sha256:3990908939ecc18f304045f3ee544bbdb68e253248fde20dc1af6829e80c03f5"},
+ {file = "pydantic_xml-2.17.2.tar.gz", hash = "sha256:0e0f2ae84518198373b593eca58bba12b08c1a7ace368dd4ed12b60babc8182d"},
+]
+
+[package.dependencies]
+pydantic = ">=2.6.0,<2.10.0b1 || >2.10.0b1"
+pydantic-core = ">=2.15.0"
+
+[package.extras]
+docs = ["Sphinx (>=5.3.0,<6.0.0)", "furo (>=2022.12.7,<2023.0.0)", "sphinx-copybutton (>=0.5.1,<0.6.0)", "sphinx_design (>=0.3.0,<0.4.0)", "toml (>=0.10.2,<0.11.0)"]
+lxml = ["lxml (>=4.9.0)"]
+
[[package]]
name = "pyee"
version = "13.0.0"
@@ -3181,4 +3201,4 @@ propcache = ">=0.2.1"
[metadata]
lock-version = "2.1"
python-versions = "~=3.13"
-content-hash = "90ea9e07b07f883138297fa9a2aac6656cc8db5d7ca25e28cb8bcd6b13027803"
+content-hash = "6f35f6e6782c7279eb97f34d47d9284126aa9a7d5c640b5a90841f559db03f3b"
diff --git a/pyproject.toml b/pyproject.toml
index 547f475..2bccb06 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -27,16 +27,17 @@ dependencies = [
"email-validator~=2.2.0",
"gitignore-parser (>=0.1.12,<0.2.0)",
"greenlet>=3.1.1,<4.0.0",
+ "htpy (>=25.7.0,<26.0.0)",
"hypercorn~=0.17",
"ldap3 (==2.10.2rc2)",
+ "pydantic-xml (>=2.17.2,<3.0.0)",
+ "pyjwt (>=2.10.1,<3.0.0)",
"python-decouple~=3.8",
"python-gnupg~=0.5",
"quart-schema[pydantic]~=0.21",
"quart-wtforms~=1.0.3",
"rich~=14.0.0",
"sqlmodel~=0.0.24",
- "pyjwt (>=2.10.1,<3.0.0)",
- "htpy (>=25.7.0,<26.0.0)",
]
[dependency-groups]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org