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/03/19 17:40:49 UTC
(tooling-trusted-release) 01/01: Improve worker code and add a basic rsync task
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
commit 4945c9992ac95b7d890376ba998768b1c25d844e
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Wed Mar 19 19:37:39 2025 +0200
Improve worker code and add a basic rsync task
---
scripts/release_path_parse.py => atr/analysis.py | 106 +++++--
atr/db/__init__.py | 20 +-
atr/ssh.py | 27 +-
atr/static/css/bootstrap.custom.css | 8 +-
atr/tasks/archive.py | 10 +-
atr/tasks/bulk.py | 5 +-
atr/tasks/license.py | 5 +-
atr/tasks/mailtest.py | 5 +-
atr/tasks/rat.py | 3 +-
atr/tasks/rsync.py | 48 ++++
atr/tasks/sbom.py | 3 +-
atr/tasks/signature.py | 2 +-
atr/tasks/task.py | 15 +-
atr/tasks/vote.py | 5 +-
atr/worker.py | 169 +++++------
bootstrap/custom.scss | 8 +-
pyproject.toml | 1 +
scripts/release_path_parse.py | 346 ++---------------------
18 files changed, 292 insertions(+), 494 deletions(-)
diff --git a/scripts/release_path_parse.py b/atr/analysis.py
similarity index 75%
copy from scripts/release_path_parse.py
copy to atr/analysis.py
index 9a434ab..54b9804 100755
--- a/scripts/release_path_parse.py
+++ b/atr/analysis.py
@@ -1,5 +1,23 @@
#!/usr/bin/env python3
+# 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 dataclasses
import os
import pathlib
import re
@@ -7,6 +25,13 @@ import signal
import sys
+@dataclasses.dataclass
+class Analysis:
+ versions: dict[str, set[str]]
+ subs: dict[str, set[str]]
+ templates: dict[str, dict[str, int]]
+
+
def architecture_pattern() -> str:
architectures = [
"cp[0-9]+-cp[0-9]+m?-[a-z0-9_]+(?:[.]manylinux[a-z0-9_]+)*",
@@ -68,6 +93,24 @@ def component_parse(i: int, component: str, size: int, elements: dict[str, str |
# elements["missing"] += 1
+def elements_update(elements: dict[str, str | None], core: str, analysis: Analysis) -> None:
+ if core not in analysis.versions:
+ analysis.versions[core] = set()
+ if core not in analysis.subs:
+ analysis.subs[core] = set()
+ if core not in analysis.templates:
+ analysis.templates[core] = {}
+
+ if elements["version"] is not None:
+ analysis.versions[core].add(elements["version"])
+ if elements["sub"] is not None:
+ analysis.subs[core].add(elements["sub"])
+ if elements["template"] is not None:
+ if elements["template"] not in analysis.templates[core]:
+ analysis.templates[core][elements["template"]] = 0
+ analysis.templates[core][elements["template"]] += 1
+
+
def extension_pattern() -> str:
# https://tableau.github.io/connector-plugin-sdk/docs/
# https://en.wikipedia.org/wiki/WAR_(file_format)
@@ -219,18 +262,27 @@ def main() -> None:
filename = sys.argv[1]
- paths = []
+ path_lines = []
with open(filename) as f:
for line in f:
- path = pathlib.Path(line.strip())
- if is_skippable(path):
- continue
- paths.append(path)
- # print(len(paths), "paths")
-
- versions: dict[str, set[str]] = {}
- subs: dict[str, set[str]] = {}
- templates: dict[str, dict[str, int]] = {}
+ path_lines.append(line.strip())
+ return perform_and_print(path_lines)
+
+
+def perform(path_lines: list[str]) -> Analysis:
+ """Perform the analysis."""
+ paths = []
+ for line in path_lines:
+ path = pathlib.Path(line.strip())
+ if is_skippable(path):
+ continue
+ paths.append(path)
+
+ analysis = Analysis(
+ versions={},
+ subs={},
+ templates={},
+ )
for path in paths:
size = len(path.parts)
elements: dict[str, str | None] = {
@@ -243,41 +295,33 @@ def main() -> None:
component_parse(i, component, size, elements)
if elements["core"] is not None:
- if elements["core"] not in versions:
- versions[elements["core"]] = set()
- if elements["core"] not in subs:
- subs[elements["core"]] = set()
- if elements["core"] not in templates:
- templates[elements["core"]] = {}
-
- if elements["version"] is not None:
- versions[elements["core"]].add(elements["version"])
- if elements["sub"] is not None:
- subs[elements["core"]].add(elements["sub"])
- if elements["template"] is not None:
- if elements["template"] not in templates[elements["core"]]:
- templates[elements["core"]][elements["template"]] = 0
- templates[elements["core"]][elements["template"]] += 1
+ elements_update(elements, elements["core"], analysis)
+
+ return analysis
+
+def perform_and_print(path_lines: list[str]) -> None:
+ """Perform the analysis and print the results."""
+ analysis = perform(path_lines)
# Prevent BrokenPipeError when piping output to other commands
signal.signal(signal.SIGPIPE, signal.SIG_DFL)
try:
- print_data(versions, subs, templates)
+ print_data(analysis)
except BrokenPipeError:
...
-def print_data(versions: dict[str, set[str]], subs: dict[str, set[str]], templates: dict[str, dict[str, int]]) -> None:
+def print_data(analysis: Analysis) -> None:
# Print the templates of all projects
- for core, version_set in sorted(versions.items()):
+ for core, version_set in sorted(analysis.versions.items()):
print("---", core, "---")
print()
if version_set:
print(" VERSIONS:", ", ".join(sorted(version_set)))
- if subs[core]:
- print(" SUBS:", ", ".join(sorted(subs[core])))
+ if analysis.subs[core]:
+ print(" SUBS:", ", ".join(sorted(analysis.subs[core])))
print()
- for template, count in sorted(templates[core].items()):
+ for template, count in sorted(analysis.templates[core].items()):
print(f" {count:3d} {template}")
print()
print()
diff --git a/atr/db/__init__.py b/atr/db/__init__.py
index 67eb157..49e5448 100644
--- a/atr/db/__init__.py
+++ b/atr/db/__init__.py
@@ -41,6 +41,7 @@ if TYPE_CHECKING:
_LOGGER: Final = logging.getLogger(__name__)
_global_async_sessionmaker: sqlalchemy.ext.asyncio.async_sessionmaker | None = None
+_global_atr_sessionmaker: sqlalchemy.ext.asyncio.async_sessionmaker | None = None
_global_sync_engine: sqlalchemy.Engine | None = None
@@ -490,8 +491,18 @@ def create_async_db_session() -> sqlalchemy.ext.asyncio.AsyncSession:
def session() -> Session:
"""Create a new asynchronous database session."""
- extensions = quart.current_app.extensions
- return util.validate_as_type(extensions["atr_db_session"](), Session)
+ global _global_atr_sessionmaker
+
+ if quart.has_app_context():
+ extensions = quart.current_app.extensions
+ return util.validate_as_type(extensions["atr_db_session"](), Session)
+ else:
+ if _global_atr_sessionmaker is None:
+ engine = create_async_engine(config.get())
+ _global_atr_sessionmaker = sqlalchemy.ext.asyncio.async_sessionmaker(
+ bind=engine, class_=Session, expire_on_commit=False
+ )
+ return util.validate_as_type(_global_atr_sessionmaker(), Session)
# FIXME: this method is deprecated and should be removed
@@ -510,7 +521,10 @@ def create_sync_db_engine() -> None:
def create_sync_db_session() -> sqlalchemy.orm.Session:
"""Create a new synchronous database session."""
global _global_sync_engine
- assert _global_sync_engine is not None
+ if _global_sync_engine is None:
+ conf = config.get()
+ sqlite_url = f"sqlite://{conf.SQLITE_DB_PATH}"
+ _global_sync_engine = sqlalchemy.create_engine(sqlite_url, echo=False)
return sqlalchemy.orm.Session(_global_sync_engine)
diff --git a/atr/ssh.py b/atr/ssh.py
index 685a7f6..a650234 100644
--- a/atr/ssh.py
+++ b/atr/ssh.py
@@ -21,6 +21,7 @@ import asyncio
import asyncio.subprocess
import logging
import os
+import uuid
from typing import Final
import aiofiles
@@ -29,6 +30,7 @@ import asyncssh
import atr.config as config
import atr.db as db
+from atr.db import models
_LOGGER: Final = logging.getLogger(__name__)
_CONFIG: Final = config.get()
@@ -167,15 +169,15 @@ async def _command_validate(process: asyncssh.SSHServerProcess) -> list[str] | N
async def _handle_client(process: asyncssh.SSHServerProcess) -> None:
"""Process client command by executing it and redirecting I/O."""
- username = process.get_extra_info("username")
- _LOGGER.info(f"Handling command for authenticated user: {username}")
+ asf_uid = process.get_extra_info("username")
+ _LOGGER.info(f"Handling command for authenticated user: {asf_uid}")
argv = await _command_validate(process)
if not argv:
return
# Create a storage directory for this user if it doesn't exist
- user_storage_dir = os.path.join(_CONFIG.STATE_DIR, "rsync-files", username)
+ user_storage_dir = os.path.join(_CONFIG.STATE_DIR, "rsync-files", asf_uid)
await aiofiles.os.makedirs(user_storage_dir, exist_ok=True)
# Set the target directory to the user's storage directory
@@ -196,8 +198,25 @@ async def _handle_client(process: asyncssh.SSHServerProcess) -> None:
# Redirect I/O between SSH process and the subprocess
await process.redirect(stdin=proc.stdin, stdout=proc.stdout, stderr=proc.stderr)
- # Wait for the process to complete and exit with its status
+ # Wait for the process to complete
exit_status = await proc.wait()
+ # Start a task to process the new files
+ async with db.session() as data:
+ async with data.begin():
+ from atr.tasks.rsync import Analyse
+
+ upload_uuid = str(uuid.uuid4()).replace("-", "")
+ project_name = "test"
+ data.add(
+ models.Task(
+ status=models.TaskStatus.QUEUED,
+ task_type="rsync_analyse",
+ task_args=Analyse(
+ asf_uid=asf_uid, upload_uuid=upload_uuid, project_name=project_name
+ ).model_dump(),
+ )
+ )
+ # Exit the SSH process with the same status
process.exit(exit_status)
except Exception as e:
diff --git a/atr/static/css/bootstrap.custom.css b/atr/static/css/bootstrap.custom.css
index 9a0aada..e69cf84 100644
--- a/atr/static/css/bootstrap.custom.css
+++ b/atr/static/css/bootstrap.custom.css
@@ -11449,10 +11449,14 @@ th {
font-weight: 525;
}
-.btn {
+small, .small, .text-muted {
+ font-variation-settings: "opsz" 14;
+}
+
+.btn, input[type=submit] {
font-family: "Inter", sans-serif;
font-size: 17px;
- font-weight: 475;
+ font-weight: 475 !important;
}
.btn:disabled {
diff --git a/atr/tasks/archive.py b/atr/tasks/archive.py
index 677cae0..7628234 100644
--- a/atr/tasks/archive.py
+++ b/atr/tasks/archive.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import asyncio
import logging
import os.path
import tarfile
@@ -23,6 +24,7 @@ from typing import Any, Final
import pydantic
import atr.tasks.task as task
+from atr.db import models
_LOGGER: Final = logging.getLogger(__name__)
@@ -34,18 +36,18 @@ class CheckIntegrity(pydantic.BaseModel):
chunk_size: int = pydantic.Field(default=4096, description="Size of chunks to read when checking the file")
-async def check_integrity(args: dict[str, Any]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+async def check_integrity(args: dict[str, Any]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Check the integrity of a .tar.gz file."""
# TODO: We should standardise the "ERROR" mechanism here in the data
# Then we can have a single task wrapper for all tasks
# TODO: We should use task.TaskError as standard, and maybe typeguard each function
data = CheckIntegrity(**args)
- task_results = task.results_as_tuple(_check_integrity_core(data.path, data.chunk_size))
+ task_results = task.results_as_tuple(await asyncio.to_thread(_check_integrity_core, data.path, data.chunk_size))
_LOGGER.info(f"Verified {data.path} and computed size {task_results[0]}")
return task.COMPLETED, None, task_results
-def check_structure(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def check_structure(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Check the structure of a .tar.gz file."""
task_results = task.results_as_tuple(_check_structure_core(*args))
_LOGGER.info(f"Verified archive structure for {args}")
@@ -73,7 +75,7 @@ def root_directory(tgz_path: str) -> str:
return root
-async def _check_integrity_core(tgz_path: str, chunk_size: int = 4096) -> int:
+def _check_integrity_core(tgz_path: str, chunk_size: int = 4096) -> int:
"""Verify a .tar.gz file and compute its uncompressed size."""
total_size = 0
diff --git a/atr/tasks/bulk.py b/atr/tasks/bulk.py
index 846c5d4..a223215 100644
--- a/atr/tasks/bulk.py
+++ b/atr/tasks/bulk.py
@@ -30,6 +30,7 @@ import sqlalchemy
import sqlalchemy.ext.asyncio
import atr.tasks.task as task
+from atr.db import models
# Configure detailed logging
_LOGGER: Final = logging.getLogger(__name__)
@@ -285,7 +286,7 @@ def database_progress_percentage_calculate(progress: tuple[int, int] | None) ->
return percentage
-async def download(args: dict[str, Any]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+async def download(args: dict[str, Any]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Download bulk package from URL."""
# Returns (status, error, result)
# This is the main task entry point, called by worker.py
@@ -302,7 +303,7 @@ async def download(args: dict[str, Any]) -> tuple[task.Status, str | None, tuple
return task.FAILED, str(e), ({"message": f"Error: {e}", "progress": 0},)
-async def download_core(args_dict: dict[str, Any]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+async def download_core(args_dict: dict[str, Any]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Download bulk package from URL."""
_LOGGER.info("Starting download_core")
try:
diff --git a/atr/tasks/license.py b/atr/tasks/license.py
index 7143f12..22ff67f 100644
--- a/atr/tasks/license.py
+++ b/atr/tasks/license.py
@@ -23,6 +23,7 @@ from typing import Any, Final
import atr.tasks.archive as archive
import atr.tasks.task as task
+from atr.db import models
_LOGGER = logging.getLogger(__name__)
@@ -168,7 +169,7 @@ under the License."""
# Tasks
-def check_files(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def check_files(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Check that the LICENSE and NOTICE files exist and are valid."""
task_results = task.results_as_tuple(_files_check_core(*args))
_LOGGER.info(f"Verified license files for {args}")
@@ -177,7 +178,7 @@ def check_files(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ..
return status, error, task_results
-def check_headers(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def check_headers(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Check that all source files have valid license headers."""
task_results = task.results_as_tuple(_headers_check_core(*args))
_LOGGER.info(f"Verified license headers for {args}")
diff --git a/atr/tasks/mailtest.py b/atr/tasks/mailtest.py
index 49ac8d1..65773e3 100644
--- a/atr/tasks/mailtest.py
+++ b/atr/tasks/mailtest.py
@@ -21,6 +21,7 @@ import os
from typing import Any, Final
import atr.tasks.task as task
+from atr.db import models
# Configure detailed logging
_LOGGER: Final = logging.getLogger(__name__)
@@ -85,7 +86,7 @@ class Args:
return args_obj
-def send(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def send(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Send a test email."""
_LOGGER.info(f"Sending with args: {args}")
try:
@@ -98,7 +99,7 @@ def send(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
return task.FAILED, str(e), tuple()
-def send_core(args_list: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def send_core(args_list: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Send a test email."""
import asyncio
diff --git a/atr/tasks/rat.py b/atr/tasks/rat.py
index 8bd74d9..42c6ab4 100644
--- a/atr/tasks/rat.py
+++ b/atr/tasks/rat.py
@@ -26,6 +26,7 @@ import atr.config as config
import atr.tasks.archive as archive
import atr.tasks.sbom as sbom
import atr.tasks.task as task
+from atr.db import models
_CONFIG: Final = config.get()
_JAVA_MEMORY_ARGS: Final[list[str]] = []
@@ -41,7 +42,7 @@ _JAVA_MEMORY_ARGS: Final[list[str]] = []
_LOGGER: Final = logging.getLogger(__name__)
-def check_licenses(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def check_licenses(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Use Apache RAT to check the licenses of the files in the artifact."""
# First argument is the artifact path
artifact_path = args[0]
diff --git a/atr/tasks/rsync.py b/atr/tasks/rsync.py
new file mode 100644
index 0000000..ddbe601
--- /dev/null
+++ b/atr/tasks/rsync.py
@@ -0,0 +1,48 @@
+# 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
+from typing import Any, Final
+
+import pydantic
+
+import atr.tasks.task as task
+from atr import analysis
+from atr.db import models
+
+_LOGGER: Final = logging.getLogger(__name__)
+
+
+class Analyse(pydantic.BaseModel):
+ """Parameters for rsync analysis."""
+
+ asf_uid: str = pydantic.Field(..., description="ASF UID of the user to rsync from")
+ upload_uuid: str = pydantic.Field(..., description="UUID of the upload to rsync")
+ project_name: str = pydantic.Field(..., description="Name of the project to rsync")
+
+
+async def analyse(args: dict[str, Any]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
+ """Analyse an rsync upload."""
+ data = Analyse(**args)
+ task_results = task.results_as_tuple(await _analyse_core(data.asf_uid, data.upload_uuid, data.project_name))
+ _LOGGER.info(f"Analyse {data.upload_uuid} for {data.project_name}")
+ return task.COMPLETED, None, task_results
+
+
+async def _analyse_core(asf_uid: str, upload_uuid: str, project_name: str) -> str:
+ """Analyse an rsync upload."""
+ return analysis.perform.__name__
diff --git a/atr/tasks/sbom.py b/atr/tasks/sbom.py
index 763d3f7..a13504d 100644
--- a/atr/tasks/sbom.py
+++ b/atr/tasks/sbom.py
@@ -23,6 +23,7 @@ from typing import Any, Final
import atr.config as config
import atr.tasks.archive as archive
import atr.tasks.task as task
+from atr.db import models
_CONFIG: Final = config.get()
_LOGGER: Final = logging.getLogger(__name__)
@@ -87,7 +88,7 @@ def archive_extract_safe(
return total_extracted
-def generate_cyclonedx(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def generate_cyclonedx(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Generate a CycloneDX SBOM for the given artifact."""
# First argument should be the artifact path
artifact_path = args[0]
diff --git a/atr/tasks/signature.py b/atr/tasks/signature.py
index 15d5c3e..f80406d 100644
--- a/atr/tasks/signature.py
+++ b/atr/tasks/signature.py
@@ -32,7 +32,7 @@ import atr.tasks.task as task
_LOGGER = logging.getLogger(__name__)
-def check(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def check(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Check a signature file."""
task_results = task.results_as_tuple(_check_core(*args))
_LOGGER.info(f"Verified {args} with result {task_results[0]}")
diff --git a/atr/tasks/task.py b/atr/tasks/task.py
index 1628de6..0c2368a 100644
--- a/atr/tasks/task.py
+++ b/atr/tasks/task.py
@@ -15,17 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-import enum
-from typing import Any, Final, Literal
+from typing import Any, Final
+from atr.db import models
-class Status(enum.Enum):
- COMPLETED = "completed"
- FAILED = "failed"
-
-
-COMPLETED: Final[Literal[Status.COMPLETED]] = Status.COMPLETED
-FAILED: Final[Literal[Status.FAILED]] = Status.FAILED
+QUEUED: Final = models.TaskStatus.QUEUED
+ACTIVE: Final = models.TaskStatus.ACTIVE
+COMPLETED: Final = models.TaskStatus.COMPLETED
+FAILED: Final = models.TaskStatus.FAILED
class Error(Exception):
diff --git a/atr/tasks/vote.py b/atr/tasks/vote.py
index c449628..1fc73b3 100644
--- a/atr/tasks/vote.py
+++ b/atr/tasks/vote.py
@@ -22,6 +22,7 @@ import os
from typing import Any, Final
import atr.tasks.task as task
+from atr.db import models
# Configure detailed logging
_LOGGER: Final = logging.getLogger(__name__)
@@ -100,7 +101,7 @@ class Args:
return args_obj
-def initiate(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def initiate(args: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Initiate a vote for a release."""
_LOGGER.info(f"Initiating vote with args: {args}")
try:
@@ -113,7 +114,7 @@ def initiate(args: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]
return task.FAILED, str(e), tuple()
-def initiate_core(args_list: list[str]) -> tuple[task.Status, str | None, tuple[Any, ...]]:
+def initiate_core(args_list: list[str]) -> tuple[models.TaskStatus, str | None, tuple[Any, ...]]:
"""Get arguments, create an email, and then send it to the recipient."""
import atr.mail
from atr.db.service import get_release_by_key_sync
diff --git a/atr/worker.py b/atr/worker.py
index 2369a20..7ba98ff 100644
--- a/atr/worker.py
+++ b/atr/worker.py
@@ -30,9 +30,10 @@ import os
import resource
import signal
import sys
+import traceback
from typing import Any, Final
-import sqlalchemy
+import sqlmodel
import atr.db as db
import atr.tasks.archive as archive
@@ -40,10 +41,12 @@ import atr.tasks.bulk as bulk
import atr.tasks.license as license
import atr.tasks.mailtest as mailtest
import atr.tasks.rat as rat
+import atr.tasks.rsync as rsync
import atr.tasks.sbom as sbom
import atr.tasks.signature as signature
import atr.tasks.task as task
import atr.tasks.vote as vote
+from atr.db import models
_LOGGER: Final = logging.getLogger(__name__)
@@ -90,118 +93,89 @@ async def _task_error_handle(task_id: int, e: Exception) -> None:
"""Handle task error by updating the database with error information."""
if isinstance(e, task.Error):
_LOGGER.error(f"Task {task_id} failed: {e.message}")
+ _LOGGER.error("".join(traceback.format_exception(e)))
result = json.dumps(e.result)
- async with db.create_async_db_session() as session:
- async with session.begin():
- await session.execute(
- sqlalchemy.text("""
- UPDATE task
- SET status = 'FAILED', completed = :now, error = :error, result = :result
- WHERE id = :task_id
- """),
- {
- "now": datetime.datetime.now(datetime.UTC),
- "task_id": task_id,
- "error": e.message,
- "result": result,
- },
- )
+
+ async with db.session() as data:
+ async with data.begin():
+ task_obj = await data.task(id=task_id).get()
+ if task_obj:
+ task_obj.status = task.FAILED
+ task_obj.completed = datetime.datetime.now(datetime.UTC)
+ task_obj.error = e.message
+ task_obj.result = result
else:
_LOGGER.error(f"Task {task_id} failed: {e}")
- async with db.create_async_db_session() as session:
- async with session.begin():
- await session.execute(
- sqlalchemy.text("""
- UPDATE task
- SET status = 'FAILED', completed = :now, error = :error
- WHERE id = :task_id
- """),
- {"now": datetime.datetime.now(datetime.UTC), "task_id": task_id, "error": str(e)},
- )
-
-
-async def _task_next_claim() -> tuple[int, str, str] | None:
+ _LOGGER.error("".join(traceback.format_exception(e)))
+
+ async with db.session() as data:
+ async with data.begin():
+ task_obj = await data.task(id=task_id).get()
+ if task_obj:
+ task_obj.status = task.FAILED
+ task_obj.completed = datetime.datetime.now(datetime.UTC)
+ task_obj.error = str(e)
+
+
+async def _task_next_claim() -> tuple[int, str, list[str] | dict[str, Any]] | None:
"""
Attempt to claim the oldest unclaimed task.
Returns (task_id, task_type, task_args) if successful.
Returns None if no tasks are available.
"""
- async with db.create_async_db_session() as session:
- async with session.begin():
- # Find and claim the oldest unclaimed task
- # We have an index on (status, added)
- result = await session.execute(
- sqlalchemy.text("""
- UPDATE task
- SET started = :now, pid = :pid, status = 'ACTIVE'
- WHERE id = (
- SELECT id FROM task
- WHERE status = 'QUEUED'
- ORDER BY added ASC LIMIT 1
- )
- AND status = 'QUEUED'
- RETURNING id, task_type, task_args
- """),
- {"now": datetime.datetime.now(datetime.UTC), "pid": os.getpid()},
+ async with db.session() as data:
+ async with data.begin():
+ # Get all queued tasks and order by added date
+ query = (
+ sqlmodel.select(models.Task)
+ .where(models.Task.status == task.QUEUED)
+ .order_by(db.validate_instrumented_attribute(models.Task.added).asc())
+ .limit(1)
)
- task = result.first()
- if task:
- task_id, task_type, task_args = task
- _LOGGER.info(f"Claimed task {task_id} ({task_type}) with args {task_args}")
- return task_id, task_type, task_args
+
+ result = await data.execute(query)
+ task_obj = result.scalar_one_or_none()
+
+ if task_obj:
+ # Update the task to mark it as active
+ task_obj.status = task.ACTIVE
+ task_obj.started = datetime.datetime.now(datetime.UTC)
+ task_obj.pid = os.getpid()
+
+ _LOGGER.info(f"Claimed task {task_obj.id} ({task_obj.task_type}) with args {task_obj.task_args}")
+ return task_obj.id, task_obj.task_type, task_obj.task_args
return None
async def _task_result_process(
- task_id: int, task_results: tuple[Any, ...], status: str = "COMPLETED", error: str | None = None
+ task_id: int, task_results: tuple[Any, ...], status: models.TaskStatus, error: str | None = None
) -> None:
"""Process and store task results in the database."""
- async with db.create_async_db_session() as session:
- result = json.dumps(task_results)
- async with session.begin():
- if status == "FAILED" and error:
- await session.execute(
- sqlalchemy.text("""
- UPDATE task
- SET status = :status, completed = :now, result = :result, error = :error
- WHERE id = :task_id
- """),
- {
- "now": datetime.datetime.now(datetime.UTC),
- "task_id": task_id,
- "result": result,
- "status": status.upper(),
- "error": error,
- },
- )
- else:
- await session.execute(
- sqlalchemy.text("""
- UPDATE task
- SET status = :status, completed = :now, result = :result
- WHERE id = :task_id
- """),
- {
- "now": datetime.datetime.now(datetime.UTC),
- "task_id": task_id,
- "result": result,
- "status": status.upper(),
- },
- )
-
-
-async def _task_process(task_id: int, task_type: str, task_args: str) -> None:
+ async with db.session() as data:
+ async with data.begin():
+ # Find the task by ID
+ task_obj = await data.task(id=task_id).get()
+ if task_obj:
+ # Update task properties
+ task_obj.status = status
+ task_obj.completed = datetime.datetime.now(datetime.UTC)
+ task_obj.result = task_results
+
+ if (status == task.FAILED) and error:
+ task_obj.error = error
+
+
+async def _task_process(task_id: int, task_type: str, task_args: list[str] | dict[str, Any]) -> None:
"""Process a claimed task."""
_LOGGER.info(f"Processing task {task_id} ({task_type}) with args {task_args}")
try:
- args = json.loads(task_args)
-
# Map task types to their handler functions
# TODO: We should use a decorator to register these automatically
dict_task_handlers = {
"verify_archive_integrity": archive.check_integrity,
"package_bulk_download": bulk.download,
+ "rsync_analyse": rsync.analyse,
}
# TODO: These are synchronous
# We plan to convert these to async dict handlers
@@ -216,34 +190,27 @@ async def _task_process(task_id: int, task_type: str, task_args: str) -> None:
"vote_initiate": vote.initiate,
}
- if isinstance(args, dict):
+ if isinstance(task_args, dict):
dict_handler = dict_task_handlers.get(task_type)
if not dict_handler:
msg = f"Unknown task type: {task_type}"
_LOGGER.error(msg)
raise Exception(msg)
- status, error, task_results = await dict_handler(args)
+ status, error, task_results = await dict_handler(task_args)
else:
list_handler = list_task_handlers.get(task_type)
if not list_handler:
msg = f"Unknown task type: {task_type}"
_LOGGER.error(msg)
raise Exception(msg)
- status, error, task_results = list_handler(args)
-
- await _task_result_process(task_id, task_results, status=status.value.upper(), error=error)
+ status, error, task_results = list_handler(task_args)
+ _LOGGER.info(f"Task {task_id} completed with status {status}, error {error}, results {task_results}")
+ await _task_result_process(task_id, task_results, status, error)
except Exception as e:
await _task_error_handle(task_id, e)
-async def _task_process_wrap(item: Any) -> tuple[Any, ...]:
- """Ensure that returned results are structured as a tuple."""
- if not isinstance(item, tuple):
- return (item,)
- return item
-
-
# Worker functions
diff --git a/bootstrap/custom.scss b/bootstrap/custom.scss
index 84e00fe..b122c1b 100644
--- a/bootstrap/custom.scss
+++ b/bootstrap/custom.scss
@@ -70,10 +70,14 @@ th {
@extend .w-25;
}
-.btn {
+small, .text-muted {
+ font-variation-settings: "opsz" 14;
+}
+
+.btn, input[type="submit"] {
font-family: $font-family-sans-serif;
font-size: 17px;
- font-weight: 475;
+ font-weight: 475 !important;
}
.btn:disabled {
diff --git a/pyproject.toml b/pyproject.toml
index 3ffeb43..ca597b1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -117,6 +117,7 @@ select = [
]
[tool.ruff.lint.per-file-ignores]
+"atr/analysis.py" = ["RUF001"]
"atr/apache.py" = ["N815", "TC003"]
"atr/db/__init__.py" = ["C901"]
"scripts/release_path_parse.py" = ["C901", "RUF001"]
diff --git a/scripts/release_path_parse.py b/scripts/release_path_parse.py
index 9a434ab..5aab22c 100755
--- a/scripts/release_path_parse.py
+++ b/scripts/release_path_parse.py
@@ -1,329 +1,21 @@
-#!/usr/bin/env python3
-
-import os
-import pathlib
-import re
-import signal
-import sys
-
-
-def architecture_pattern() -> str:
- architectures = [
- "cp[0-9]+-cp[0-9]+m?-[a-z0-9_]+(?:[.]manylinux[a-z0-9_]+)*",
- "pp[0-9]+-pypy[0-9]+_pp[0-9]+-[a-z0-9_]+(?:[.]manylinux[a-z0-9_]+)?",
- "darwin(?:-unknown)?-(?:aarch64|amd64|arm64|64bit|arm64bit|x64)",
- "Linux-CentOS[0-9]+",
- "Linux-Ubuntu[0-9]+",
- "linux(?:-glibc|musl|unknown)?-(?:aarch64|amd64|arm64|64bit|arm64bit|x64)",
- "linux.gtk.x86_64",
- "mac(?:os|OS)?(?:-unknown)?-(?:aarch64|amd64|arm64|64bit|arm64bit|x64)",
- "macos.cocoa.x86_64",
- "osx(?:-unknown)?-(?:aarch64|amd64|arm64|64bit|arm64bit|x64)",
- "py2.py3-none-any",
- "py3-none-any",
- "win32.win32.x86_64",
- "windows(?:-unknown)?-(?:aarch64|amd64|arm64|64bit|arm64bit|x64)",
- "x86_64(?:-noavx2)?",
- "(?:x64|x86)-windows-staticaarch64",
- "amd64",
- "arm",
- "Darwin",
- "Linux_x86",
- "linux",
- "MacOS_x86-64",
- "macosx?",
- "noarch",
- "Win_x86",
- "win(?:dows)?",
- ]
- return "(" + "|".join(architectures) + ")(?=[_.-])"
-
-
-def component_parse(i: int, component: str, size: int, elements: dict[str, str | None]) -> None:
- if i == 0:
- # CORE
- # Never starts with "apache-"
- elements["core"] = component
- elif (i == 1) and (size == 2):
- elements["template"] = filename_parse(component, elements)
- elif i == 1:
- # SUB or VERSION
- # TODO: Check total depth to give an indication of SUB?
- if is_version(component):
- elements["version"] = version_parse(component, elements)
- else:
- elements["sub"] = component
- elif (i == 2) and (size == 3):
- # CORE/VERSION/FILENAME
- elements["template"] = filename_parse(component, elements)
- elif (i == 2) and (size == 4):
- # VERSION
- elements["version"] = version_parse(component, elements)
- elif (i == 3) and (size == 4):
- # CORE/VERSION/SUB/FILENAME
- elements["template"] = filename_parse(component, elements)
- elif i == (size - 1):
- # FILENAME, but more deeply nested
- elements["template"] = filename_parse(component, elements)
- # elements["missing"] += 1
-
-
-def extension_pattern() -> str:
- # https://tableau.github.io/connector-plugin-sdk/docs/
- # https://en.wikipedia.org/wiki/WAR_(file_format)
- # https://learn.microsoft.com/en-us/visualstudio/extensibility/anatomy-of-a-vsix-package?view=vs-2022
- # What's the status of "pom"?
- # We've included "sh", so perhaps we should include "patch"
- archives = [
- "bin",
- "crate",
- "deb",
- "dmg",
- "exe",
- "far",
- "gem",
- "jar.pack.gz",
- "jar",
- "msi",
- "nar",
- "nbm",
- "snupkg",
- "nupkg",
- "pkg",
- "pom",
- "rar",
- "rpm",
- "sh",
- "slingosgifeature",
- "taco",
- "tar.bz2",
- "tar.gz",
- "tar.xz",
- "tar",
- "tgz",
- "vsix",
- "war",
- "whl",
- "zip",
- ]
- # "mds" is used in ozone
- # "SHA256" and "SHA512" are used in ranger
- # "MD5" is used in samza
- # "asc.asc" is used in felix
- metadata = [
- "asc.asc",
- "asc.md5",
- "asc.sha1",
- "asc.sha256",
- "asc.sha512",
- "sha512.asc",
- "sha512.md5",
- "sha512.sha1",
- "sha512.sha512",
- "asc",
- "MD5",
- "md5",
- "mds",
- "prov",
- "sh1",
- "sha1",
- "sha256",
- "SHA256",
- "SHA512",
- "sha512sum",
- "sha512",
- "sha",
- "sig",
- ]
- patterns = []
- for archive in archives:
- for metadatum in metadata:
- patterns.append(f"[.]{archive}[.]{metadatum}")
- # This must come after the metadata patterns
- patterns.append(f"[.]{archive}")
- return "(" + "|".join(patterns) + ")$"
-
-
-def filename_parse(filename: str, elements: dict[str, str | None]) -> str:
- filename = re.sub(r"apache(?=[_.-])", "α", filename)
- # TODO: -incubating
- # There is no standard position for -incubating
- if elements["sub"]:
- # Replace SUB before CORE because CORE may contain SUB
- filename = re.sub(elements["sub"] + r"(?=[_.-])", "σ", filename)
- if elements["core"]:
- filename = re.sub(elements["core"] + r"(?=[_.-])", "κ", filename)
- if elements["version"]:
- filename = re.sub(elements["version"] + r"(?=[_.-])", "β", filename)
- filename = re.sub(variant_pattern(), "ρ", filename)
- filename = re.sub(r"[0-9]+[.][0-9]+(?:[.][0-9]+(?:[.][0-9]+)?)?(?=[_.-])", "τ", filename)
- filename = re.sub(architecture_pattern(), "ι", filename)
- filename = re.sub(extension_pattern(), ".ε", filename)
- if "LABEL_MODE" in os.environ:
- filename = re.sub(r"(?<=-)[a-z]+[0-9]*(?:-[a-z]+[0-9]*)*(?=-)", "λ", filename)
-
- filename = filename.replace("α", "ASF")
- filename = filename.replace("σ", "SUB")
- filename = filename.replace("β", "VERSION")
- filename = filename.replace("κ", "CORE")
- filename = filename.replace("ρ", "VARIANT")
- filename = filename.replace("τ", "TAG")
- filename = filename.replace("ι", "ARCH")
- filename = filename.replace("ε", "EXT")
- if "LABEL_MODE" in os.environ:
- filename = filename.replace("λ", "LABEL")
- return filename
-
-
-def is_skippable(path: pathlib.Path) -> bool:
- if len(path.parts) < 2:
- return True
- if path.parts[0] == "META":
- return True
- # "KEYS", "LICENSE", "NOTICE", "README"...
- if "." not in path.name:
- return True
- if path.name in {".htaccess"}:
- return True
- # .license is used in high volume in netbeans
- skippable = [
- ".bak",
- ".css",
- ".gif",
- ".html",
- ".json",
- ".license",
- ".md",
- ".pdf",
- ".png",
- ".temp",
- ".tmp",
- ".txt",
- ".xml",
- ".yaml",
- ]
- for suffix in skippable:
- if suffix in path.suffixes:
- return True
- return False
-
-
-def is_version(component: str) -> bool:
- return component[:1].isdigit() or ("." in component)
-
-
-def main() -> None:
- if len(sys.argv) != 2:
- print(f"Usage: {sys.argv[0]} FILENAME")
- sys.exit(1)
-
- filename = sys.argv[1]
-
- paths = []
- with open(filename) as f:
- for line in f:
- path = pathlib.Path(line.strip())
- if is_skippable(path):
- continue
- paths.append(path)
- # print(len(paths), "paths")
-
- versions: dict[str, set[str]] = {}
- subs: dict[str, set[str]] = {}
- templates: dict[str, dict[str, int]] = {}
- for path in paths:
- size = len(path.parts)
- elements: dict[str, str | None] = {
- "core": None,
- "version": None,
- "sub": None,
- "template": None,
- }
- for i, component in enumerate(path.parts):
- component_parse(i, component, size, elements)
-
- if elements["core"] is not None:
- if elements["core"] not in versions:
- versions[elements["core"]] = set()
- if elements["core"] not in subs:
- subs[elements["core"]] = set()
- if elements["core"] not in templates:
- templates[elements["core"]] = {}
-
- if elements["version"] is not None:
- versions[elements["core"]].add(elements["version"])
- if elements["sub"] is not None:
- subs[elements["core"]].add(elements["sub"])
- if elements["template"] is not None:
- if elements["template"] not in templates[elements["core"]]:
- templates[elements["core"]][elements["template"]] = 0
- templates[elements["core"]][elements["template"]] += 1
-
- # Prevent BrokenPipeError when piping output to other commands
- signal.signal(signal.SIGPIPE, signal.SIG_DFL)
- try:
- print_data(versions, subs, templates)
- except BrokenPipeError:
- ...
-
-
-def print_data(versions: dict[str, set[str]], subs: dict[str, set[str]], templates: dict[str, dict[str, int]]) -> None:
- # Print the templates of all projects
- for core, version_set in sorted(versions.items()):
- print("---", core, "---")
- print()
- if version_set:
- print(" VERSIONS:", ", ".join(sorted(version_set)))
- if subs[core]:
- print(" SUBS:", ", ".join(sorted(subs[core])))
- print()
- for template, count in sorted(templates[core].items()):
- print(f" {count:3d} {template}")
- print()
- print()
- sys.stdout.flush()
-
-
-def variant_pattern() -> str:
- # Should perhaps not include javadoc
- # app
- # doc
- # docs
- # example
- # markdown
- # nodeps
- # release
- # sdk
- # tests
- variants = [
- "binary-assembly",
- "binary",
- "bin",
- "dist",
- "install_[a-z][a-z](?:-[A-Z][A-Z])?",
- "javadoc",
- "langpack_[a-z][a-z](?:-[A-Z][A-Z])?",
- "lib-debug",
- "lib",
- "pkg",
- "source-release",
- "sources",
- "source",
- "src",
- ]
- # .bin can also be an EXT
- # For example in opennlp
- # Which is why we do (?<=[_-])
- return "(?<=[_-])(" + "|".join(variants) + ")(?=[_.-])"
-
-
-def version_parse(version: str, elements: dict[str, str | None]) -> str:
- if elements["core"] is None:
- return version
- if version.startswith(elements["core"] + "-"):
- return version[len(elements["core"]) + 1 :]
- return version
-
+# 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 atr.analysis as analysis
if __name__ == "__main__":
- main()
+ analysis.main()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org