You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by ar...@apache.org on 2026/01/29 15:04:43 UTC
(tooling-trusted-releases) branch pending_dist_changes created (now 8ed69eb)
This is an automated email from the ASF dual-hosted git repository.
arm pushed a change to branch pending_dist_changes
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
at 8ed69eb #216 - Scheduled task for pending distributions, add created_by to dist table.
This branch includes the following new commits:
new d7d8967 #216 - Add pending distribution status and background task to check it. Refactor some of the distribution logic out to shared module and some of shared module to precent circular references.
new 8ed69eb #216 - Scheduled task for pending distributions, add created_by to dist table.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-trusted-releases) 02/02: #216 - Scheduled task for pending distributions, add created_by to dist table.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arm pushed a commit to branch pending_dist_changes
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit 8ed69eb50c6941792f66447bf63cdebcfdff9a3e
Author: Alastair McFarlane <ar...@apache.org>
AuthorDate: Thu Jan 29 15:04:32 2026 +0000
#216 - Scheduled task for pending distributions, add created_by to dist table.
---
atr/get/distribution.py | 6 ++++-
atr/models/sql.py | 1 +
atr/server.py | 3 +++
atr/storage/writers/distributions.py | 2 ++
atr/tasks/__init__.py | 1 +
atr/tasks/distribution.py | 28 +++++++++++++++------
migrations/versions/0043_2026.01.29_d7d89670.py | 33 +++++++++++++++++++++++++
7 files changed, 65 insertions(+), 9 deletions(-)
diff --git a/atr/get/distribution.py b/atr/get/distribution.py
index 7bc9162..4bbf922 100644
--- a/atr/get/distribution.py
+++ b/atr/get/distribution.py
@@ -17,6 +17,7 @@
from collections.abc import Sequence
import asfquart.base as base
+import htpy
import atr.blueprints.get as get
import atr.db as db
@@ -77,12 +78,15 @@ async def list_get(session: web.Committer, project_name: str, version_name: str)
## Distributions
block.h2["Distributions"]
for dist in distributions:
+ title_extra = []
+ if dist.pending:
+ title_extra.append(htpy.small(".text-muted")[" (this distribution is being verified by ATR)"])
### Platform package version
block.h3(
# Cannot use "#id" here, because the ID contains "."
# If an ID contains ".", htm parses that as a class
id=f"distribution-{dist.identifier}"
- )[dist.title]
+ )[dist.title, *title_extra]
tbody = htm.tbody[
shared.distribution.html_tr("Release name", dist.release_name),
shared.distribution.html_tr("Platform", dist.platform.value.name),
diff --git a/atr/models/sql.py b/atr/models/sql.py
index e830083..0b8d8a4 100644
--- a/atr/models/sql.py
+++ b/atr/models/sql.py
@@ -976,6 +976,7 @@ class Distribution(sqlmodel.SQLModel, table=True):
upload_date: datetime.datetime | None = sqlmodel.Field(default=None)
api_url: str | None = sqlmodel.Field(default=None)
web_url: str | None = sqlmodel.Field(default=None)
+ created_by: str | None = sqlmodel.Field(default=None)
# The API response can be huge, e.g. from npm
# So we do not store it in the database
# api_response: Any = sqlmodel.Field(sa_column=sqlalchemy.Column(sqlalchemy.JSON))
diff --git a/atr/server.py b/atr/server.py
index 1ef0d60..70438d1 100644
--- a/atr/server.py
+++ b/atr/server.py
@@ -802,6 +802,9 @@ async def _register_recurrent_tasks() -> None:
await asyncio.sleep(60)
workflow = await tasks.workflow_update(asf_uid="system", schedule_next=True)
log.info(f"Scheduled workflow status update with ID {workflow.id}")
+ await asyncio.sleep(60)
+ dist_check = await tasks.distribution_status_check(asf_uid="system", schedule_next=True)
+ log.info(f"Scheduled distribution status update with ID {dist_check.id}")
except Exception as e:
log.exception(f"Failed to schedule recurrent tasks: {e!s}")
diff --git a/atr/storage/writers/distributions.py b/atr/storage/writers/distributions.py
index 312a471..7a4b683 100644
--- a/atr/storage/writers/distributions.py
+++ b/atr/storage/writers/distributions.py
@@ -160,6 +160,7 @@ class CommitteeMember(CommitteeParticipant):
upload_date=upload_date,
api_url=api_url,
web_url=web_url,
+ created_by=self.__asf_uid,
)
if existing is None:
self.__data.add(dist)
@@ -271,6 +272,7 @@ class CommitteeMember(CommitteeParticipant):
existing.upload_date = upload_date
existing.api_url = api_url
existing.web_url = web_url
+ existing.created_by = self.__asf_uid
await self.__data.commit()
return existing
return None
diff --git a/atr/tasks/__init__.py b/atr/tasks/__init__.py
index ea613bb..8ba7fd3 100644
--- a/atr/tasks/__init__.py
+++ b/atr/tasks/__init__.py
@@ -71,6 +71,7 @@ async def clear_scheduled(caller_data: db.Session | None = None) -> None:
[
sql.TaskType.METADATA_UPDATE,
sql.TaskType.WORKFLOW_STATUS,
+ sql.TaskType.DISTRIBUTION_STATUS,
]
),
via(sql.Task.status) == sql.TaskStatus.QUEUED,
diff --git a/atr/tasks/distribution.py b/atr/tasks/distribution.py
index 541f918..36caabf 100644
--- a/atr/tasks/distribution.py
+++ b/atr/tasks/distribution.py
@@ -39,10 +39,12 @@ class DistributionStatusCheckArgs(schema.Strict):
@checks.with_model(DistributionStatusCheckArgs)
async def status_check(args: DistributionStatusCheckArgs, *, task_id: int | None = None) -> results.Results | None:
+ log.info("Checking pending recorded distributions")
dists = []
async with db.session() as data:
dists = await data.distribution(pending=True, _with_release=True, _with_release_project=True).all()
for dist in dists:
+ name = f"{dist.platform} {dist.owner_namespace} {dist.package} {dist.version}"
dd = distribution.Data(
platform=dist.platform,
owner_namespace=dist.owner_namespace,
@@ -50,30 +52,40 @@ async def status_check(args: DistributionStatusCheckArgs, *, task_id: int | None
version=dist.version,
details=False,
)
+ if not dist.created_by:
+ log.warning(f"Distribution {name} has no creator, skipping")
+ continue
+ if not dist.release.project.committee_name:
+ log.warning(f"Distribution {name} has no committee, skipping")
+ continue
try:
- async with storage.write_as_committee_member(str(dist.release.project.committee_name), args.asf_uid) as w:
+ async with storage.write_as_committee_member(dist.release.project.committee_name, dist.created_by) as w:
if dist.retries >= _RETRY_LIMIT:
await w.distributions.delete_distribution(
dist.release_name, dist.platform, dist.owner_namespace, dist.package, dist.version
)
- name = f"{dist.platform} {dist.owner_namespace} {dist.package} {dist.version}"
log.error(f"Distribution {name} failed {_RETRY_LIMIT} times, skipping")
continue
+ log.warning(f"Retrying distribution {name}")
await w.distributions.record_from_data(
dist.release_name,
dist.staging,
dd,
+ allow_retries=True,
)
except storage.AccessError as e:
msg = f"Failed to record distribution: {e}"
log.error(msg)
raise RuntimeError(msg)
- if args.next_schedule_seconds:
- next_schedule = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=args.next_schedule_seconds)
- await tasks.distribution_status_check(args.asf_uid, schedule=next_schedule, schedule_next=True)
- log.info(
- f"Scheduled next workflow status update for: {next_schedule.strftime('%Y-%m-%d %H:%M:%S')}",
- )
+ finally:
+ if args.next_schedule_seconds:
+ next_schedule = datetime.datetime.now(datetime.UTC) + datetime.timedelta(
+ seconds=args.next_schedule_seconds
+ )
+ await tasks.distribution_status_check(args.asf_uid, schedule=next_schedule, schedule_next=True)
+ log.info(
+ f"Scheduled next workflow status update for: {next_schedule.strftime('%Y-%m-%d %H:%M:%S')}",
+ )
return results.DistributionStatusCheck(
kind="distribution_status",
)
diff --git a/migrations/versions/0043_2026.01.29_d7d89670.py b/migrations/versions/0043_2026.01.29_d7d89670.py
new file mode 100644
index 0000000..8342377
--- /dev/null
+++ b/migrations/versions/0043_2026.01.29_d7d89670.py
@@ -0,0 +1,33 @@
+"""column for distribution creator
+
+Revision ID: 0043_2026.01.29_d7d89670
+Revises: 0042_2026.01.28_3e434625
+Create Date: 2026-01-29 13:56:34.371692+00:00
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# Revision identifiers, used by Alembic
+revision: str = "0043_2026.01.29_d7d89670"
+down_revision: str | None = "0042_2026.01.28_3e434625"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+ # ### commands auto generated by Alembic - please adjust! ###
+ with op.batch_alter_table("distribution", schema=None) as batch_op:
+ batch_op.add_column(sa.Column("created_by", sa.String(), nullable=True))
+
+ # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+ # ### commands auto generated by Alembic - please adjust! ###
+ with op.batch_alter_table("distribution", schema=None) as batch_op:
+ batch_op.drop_column("created_by")
+
+ # ### end Alembic commands ###
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-trusted-releases) 01/02: #216 - Add pending distribution status and background task to check it. Refactor some of the distribution logic out to shared module and some of shared module to precent circular references.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arm pushed a commit to branch pending_dist_changes
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit d7d89670f92ddb54e81104d66e55d19f50177702
Author: Alastair McFarlane <ar...@apache.org>
AuthorDate: Thu Jan 29 11:48:53 2026 +0000
#216 - Add pending distribution status and background task to check it. Refactor some of the distribution logic out to shared module and some of shared module to precent circular references.
---
atr/db/__init__.py | 9 +
atr/get/checks.py | 2 +-
atr/get/compose.py | 2 +-
atr/models/__init__.py | 4 +-
atr/models/results.py | 9 +-
atr/models/sql.py | 5 +-
atr/post/distribution.py | 1 +
atr/shared/__init__.py | 215 +-------------
atr/shared/distribution.py | 247 ++++++++++++++++
atr/shared/{__init__.py => web.py} | 70 +----
atr/storage/writers/distributions.py | 360 +++++-------------------
atr/tasks/__init__.py | 30 ++
atr/tasks/distribution.py | 79 ++++++
migrations/versions/0042_2026.01.28_3e434625.py | 31 ++
14 files changed, 489 insertions(+), 575 deletions(-)
diff --git a/atr/db/__init__.py b/atr/db/__init__.py
index 4ff46d7..80ebbc5 100644
--- a/atr/db/__init__.py
+++ b/atr/db/__init__.py
@@ -289,6 +289,9 @@ class Session(sqlalchemy.ext.asyncio.AsyncSession):
owner_namespace: Opt[str] = NOT_SET,
package: Opt[str] = NOT_SET,
version: Opt[str] = NOT_SET,
+ pending: Opt[bool] = NOT_SET,
+ _with_release: bool = False,
+ _with_release_project: bool = False,
) -> Query[sql.Distribution]:
query = sqlmodel.select(sql.Distribution)
if is_defined(release_name):
@@ -301,6 +304,12 @@ class Session(sqlalchemy.ext.asyncio.AsyncSession):
query = query.where(sql.Distribution.package == package)
if is_defined(version):
query = query.where(sql.Distribution.version == version)
+ if is_defined(pending):
+ query = query.where(sql.Distribution.pending == pending)
+ if _with_release_project:
+ query = query.options(joined_load_nested(sql.Distribution.release, sql.Release.project))
+ elif _with_release:
+ query = query.options(joined_load(sql.Distribution.release))
return Query(self, query)
async def execute_query(self, query: sqlalchemy.sql.expression.Executable) -> sqlalchemy.engine.Result:
diff --git a/atr/get/checks.py b/atr/get/checks.py
index 1781b2f..df36f9e 100644
--- a/atr/get/checks.py
+++ b/atr/get/checks.py
@@ -162,7 +162,7 @@ async def selected_revision(
ongoing_count = await interaction.tasks_ongoing(project_name, version_name, revision_number)
- checks_summary_elem = shared._render_checks_summary(info, project_name, version_name)
+ checks_summary_elem = shared.web._render_checks_summary(info, project_name, version_name)
checks_summary_html = str(checks_summary_elem) if checks_summary_elem else ""
delete_file_forms: dict[str, str] = {}
diff --git a/atr/get/compose.py b/atr/get/compose.py
index 304f246..6e40087 100644
--- a/atr/get/compose.py
+++ b/atr/get/compose.py
@@ -39,4 +39,4 @@ async def selected(session: web.Committer, project_name: str, version_name: str)
).demand(base.ASFQuartException("Release does not exist", errorcode=404))
if release.phase != sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT:
return await mapping.release_as_redirect(session, release)
- return await shared.check(session, release)
+ return await shared.web.check(session, release)
diff --git a/atr/models/__init__.py b/atr/models/__init__.py
index 52c3cad..daca400 100644
--- a/atr/models/__init__.py
+++ b/atr/models/__init__.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-from . import api, distribution, helpers, results, schema, sql, tabulate, validation
+from . import api, basic, distribution, helpers, results, schema, sql, tabulate, validation
# If we use .__name__, pyright gives a warning
-__all__ = ["api", "distribution", "helpers", "results", "schema", "sql", "tabulate", "validation"]
+__all__ = ["api", "basic", "distribution", "helpers", "results", "schema", "sql", "tabulate", "validation"]
diff --git a/atr/models/results.py b/atr/models/results.py
index 1ecb4d9..eeb73b9 100644
--- a/atr/models/results.py
+++ b/atr/models/results.py
@@ -22,6 +22,12 @@ import pydantic
from . import schema
+class DistributionStatusCheck(schema.Strict):
+ """Result of the task to check pending distribution statuses."""
+
+ kind: Literal["distribution_status"] = schema.Field(alias="kind")
+
+
class DistributionWorkflow(schema.Strict):
"""Result of the task to run a Github workflow."""
@@ -228,7 +234,8 @@ class MetadataUpdate(schema.Strict):
Results = Annotated[
- DistributionWorkflow
+ DistributionStatusCheck
+ | DistributionWorkflow
| DistributionWorkflowStatus
| HashingCheck
| MessageSend
diff --git a/atr/models/sql.py b/atr/models/sql.py
index 9887711..e830083 100644
--- a/atr/models/sql.py
+++ b/atr/models/sql.py
@@ -190,6 +190,7 @@ class TaskStatus(str, enum.Enum):
class TaskType(str, enum.Enum):
+ DISTRIBUTION_STATUS = "distribution_status"
DISTRIBUTION_WORKFLOW = "distribution_workflow"
HASHING_CHECK = "hashing_check"
KEYS_IMPORT_FILE = "keys_import_file"
@@ -970,8 +971,10 @@ class Distribution(sqlmodel.SQLModel, table=True):
package: str = sqlmodel.Field(primary_key=True, index=True)
version: str = sqlmodel.Field(primary_key=True, index=True)
staging: bool = sqlmodel.Field(default=False)
+ pending: bool = sqlmodel.Field(default=False)
+ retries: int = sqlmodel.Field(default=0)
upload_date: datetime.datetime | None = sqlmodel.Field(default=None)
- api_url: str
+ api_url: str | None = sqlmodel.Field(default=None)
web_url: str | None = sqlmodel.Field(default=None)
# The API response can be huge, e.g. from npm
# So we do not store it in the database
diff --git a/atr/post/distribution.py b/atr/post/distribution.py
index 706bafa..438a16a 100644
--- a/atr/post/distribution.py
+++ b/atr/post/distribution.py
@@ -176,6 +176,7 @@ async def record_form_process_page(
release_name=release.name,
staging=staging,
dd=dd,
+ allow_retries=True,
)
except storage.AccessError as e:
# Instead of calling record_form_page_new, redirect with error message
diff --git a/atr/shared/__init__.py b/atr/shared/__init__.py
index 4c9e400..4090d59 100644
--- a/atr/shared/__init__.py
+++ b/atr/shared/__init__.py
@@ -15,18 +15,8 @@
# specific language governing permissions and limitations
# under the License.
-from typing import TYPE_CHECKING, Final
+from typing import Final
-import htpy
-
-import atr.db as db
-import atr.db.interaction as interaction
-import atr.form as form
-import atr.get as get
-import atr.htm as htm
-import atr.models.results as results
-import atr.models.sql as sql
-import atr.post as post
import atr.shared.announce as announce
import atr.shared.distribution as distribution
import atr.shared.draft as draft
@@ -45,15 +35,7 @@ import atr.shared.upload as upload
import atr.shared.user as user
import atr.shared.vote as vote
import atr.shared.voting as voting
-import atr.storage as storage
-import atr.storage.types as types
-import atr.template as template
-import atr.util as util
-import atr.web as web
-
-if TYPE_CHECKING:
- from collections.abc import Sequence
-
+import atr.shared.web as web
# | 1 | RSA (Encrypt or Sign) [HAC] |
# | 2 | RSA Encrypt-Only [HAC] |
@@ -81,201 +63,9 @@ algorithms: Final[dict[int, str]] = {
}
-async def check(
- session: web.Committer | None,
- release: sql.Release,
- task_mid: str | None = None,
- vote_form: htm.Element | None = None,
- resolve_form: htm.Element | None = None,
- archive_url: str | None = None,
- vote_task: sql.Task | None = None,
- can_vote: bool = False,
- can_resolve: bool = False,
-) -> web.WerkzeugResponse | str:
- base_path = util.release_directory(release)
-
- # TODO: This takes 180ms for providers
- # We could cache it
- paths = [path async for path in util.paths_recursive(base_path)]
- paths.sort()
-
- async with storage.read(session) as read:
- ragp = read.as_general_public()
- info = await ragp.releases.path_info(release, paths)
-
- user_ssh_keys: Sequence[sql.SSHKey] = []
- asf_id: str | None = None
- server_domain: str | None = None
- server_host: str | None = None
-
- if session is not None:
- asf_id = session.uid
- server_domain = session.app_host.split(":", 1)[0]
- server_host = session.app_host
- async with db.session() as data:
- user_ssh_keys = await data.ssh_key(asf_uid=session.uid).all()
-
- # Get the number of ongoing tasks for the current revision
- ongoing_tasks_count = 0
- match await interaction.latest_info(release.project.name, release.version):
- case (revision_number, revision_editor, revision_timestamp):
- ongoing_tasks_count = await interaction.tasks_ongoing(
- release.project.name,
- release.version,
- revision_number,
- )
- case None:
- revision_number = None
- revision_editor = None
- revision_timestamp = None
-
- delete_form = form.render(
- model_cls=form.Empty,
- action=util.as_url(get.compose.selected, project_name=release.project.name, version_name=release.version),
- submit_label="Delete this draft",
- submit_classes="btn btn-danger",
- empty=True,
- confirm="Are you sure you want to delete this draft? This cannot be undone.",
- )
-
- delete_file_forms: dict[str, htm.Element] = {}
- for path in paths:
- delete_file_forms[str(path)] = form.render(
- model_cls=draft.DeleteFileForm,
- action=util.as_url(post.draft.delete_file, project_name=release.project.name, version_name=release.version),
- form_classes=".d-inline-block.m-0",
- submit_classes="btn-sm btn-outline-danger",
- submit_label="Delete",
- empty=True,
- defaults={"file_path": str(path)},
- # TODO: Add a static check for the confirm syntax
- confirm=(
- "Are you sure you want to delete this file? "
- "This will also delete any associated metadata files. "
- "This cannot be undone."
- ),
- )
-
- empty_form = form.render(
- model_cls=form.Empty,
- action=util.as_url(post.draft.fresh, project_name=release.project.name, version_name=release.version),
- submit_label="Restart all checks",
- submit_classes="btn btn-primary",
- )
-
- vote_task_warnings = _warnings_from_vote_result(vote_task)
- has_files = await util.has_files(release)
-
- has_any_errors = any(info.errors.get(path, []) for path in paths) if info else False
- strict_checking = release.project.policy_strict_checking
- strict_checking_errors = strict_checking and has_any_errors
-
- checks_summary_html = _render_checks_summary(info, release.project.name, release.version)
-
- return await template.render(
- "check-selected.html",
- project_name=release.project.name,
- version_name=release.version,
- release=release,
- paths=paths,
- info=info,
- revision_editor=revision_editor,
- revision_time=revision_timestamp,
- revision_number=revision_number,
- ongoing_tasks_count=ongoing_tasks_count,
- delete_form=delete_form,
- delete_file_forms=delete_file_forms,
- asf_id=asf_id,
- server_domain=server_domain,
- server_host=server_host,
- user_ssh_keys=user_ssh_keys,
- format_datetime=util.format_datetime,
- models=sql,
- task_mid=task_mid,
- vote_form=vote_form,
- vote_task=vote_task,
- archive_url=archive_url,
- vote_task_warnings=vote_task_warnings,
- empty_form=empty_form,
- csrf_input=str(form.csrf_input()),
- resolve_form=resolve_form,
- has_files=has_files,
- strict_checking_errors=strict_checking_errors,
- can_vote=can_vote,
- can_resolve=can_resolve,
- checks_summary_html=checks_summary_html,
- )
-
-
-def _checker_display_name(checker: str) -> str:
- return checker.removeprefix("atr.tasks.checks.").replace("_", " ").replace(".", " ").title()
-
-
-def _render_checks_summary(info: types.PathInfo | None, project_name: str, version_name: str) -> htm.Element | None:
- if (info is None) or (not info.checker_stats):
- return None
-
- card = htm.Block(htm.div, classes=".card.mb-4")
- card.div(".card-header")[htpy.h5(".mb-0")["Checks summary"]]
-
- body = htm.Block(htm.div, classes=".card-body")
- for i, stat in enumerate(info.checker_stats):
- stripe_class = ".atr-stripe-odd" if ((i % 2) == 0) else ".atr-stripe-even"
- details = htm.Block(htm.details, classes=f".mb-0.p-2{stripe_class}")
-
- summary_content: list[htm.Element | str] = []
- if stat.warning_count > 0:
- summary_content.append(htpy.span(".badge.bg-warning.text-dark.me-2")[str(stat.warning_count)])
- if stat.failure_count > 0:
- summary_content.append(htpy.span(".badge.bg-danger.me-2")[str(stat.failure_count)])
- summary_content.append(htpy.strong[_checker_display_name(stat.checker)])
-
- details.summary[*summary_content]
-
- files_div = htm.Block(htm.div, classes=".mt-2.atr-checks-files")
- all_files = set(stat.failure_files.keys()) | set(stat.warning_files.keys())
- for file_path in sorted(all_files):
- report_url = f"/report/{project_name}/{version_name}/{file_path}"
- error_count = stat.failure_files.get(file_path, 0)
- warning_count = stat.warning_files.get(file_path, 0)
-
- file_content: list[htm.Element | str] = []
- if error_count > 0:
- file_content.append(htpy.span(".badge.bg-danger.me-2")[util.plural(error_count, "error")])
- if warning_count > 0:
- file_content.append(
- htpy.span(".badge.bg-warning.text-dark.me-2")[util.plural(warning_count, "warning")]
- )
- file_content.append(htpy.a(href=report_url)[htpy.strong[htpy.code[file_path]]])
-
- files_div.div[*file_content]
-
- details.append(files_div.collect())
- body.append(details.collect())
-
- card.append(body.collect())
- return card.collect()
-
-
-def _warnings_from_vote_result(vote_task: sql.Task | None) -> list[str]:
- # TODO: Replace this with a schema.Strict model
- # But we'd still need to do some of this parsing and validation
- # We should probably rethink how to send data through tasks
-
- if (not vote_task) or (not vote_task.result):
- return ["No vote task result found."]
-
- vote_task_result = vote_task.result
- if not isinstance(vote_task_result, results.VoteInitiate):
- return ["Vote task result is not a results.VoteInitiate instance."]
-
- return vote_task_result.mail_send_warnings
-
-
__all__ = [
"algorithms",
"announce",
- "check",
"distribution",
"draft",
"finish",
@@ -293,4 +83,5 @@ __all__ = [
"user",
"vote",
"voting",
+ "web",
]
diff --git a/atr/shared/distribution.py b/atr/shared/distribution.py
index 743c781..e0af07a 100644
--- a/atr/shared/distribution.py
+++ b/atr/shared/distribution.py
@@ -17,15 +17,20 @@
from __future__ import annotations
+import datetime
import enum
+import aiohttp
import pydantic
import atr.db as db
import atr.form as form
import atr.htm as htm
+import atr.models.basic as basic
import atr.models.distribution as distribution
import atr.models.sql as sql
+import atr.util as util
+from atr.storage import outcome
class DistributionPlatform(enum.Enum):
@@ -131,6 +136,25 @@ def html_tr_a(label: str, value: str | None) -> htm.Element:
return htm.tr[htm.th[label], htm.td[htm.a(href=value)[value] if value else "-"]]
+async def json_from_distribution_platform(
+ api_url: str, platform: sql.DistributionPlatform, version: str
+) -> outcome.Outcome[basic.JSON]:
+ try:
+ async with util.create_secure_session() as session:
+ async with session.get(api_url) as response:
+ response.raise_for_status()
+ response_json = await response.json()
+ result = basic.as_json(response_json)
+ except aiohttp.ClientError as e:
+ return outcome.Error(e)
+ match platform:
+ case sql.DistributionPlatform.NPM | sql.DistributionPlatform.NPM_SCOPED:
+ if version not in distribution.NpmResponse.model_validate(result).time:
+ e = RuntimeError(f"Version '{version}' not found")
+ return outcome.Error(e)
+ return outcome.Result(result)
+
+
async def release_validated(
project: str, version: str, committee: bool = False, staging: bool | None = None, release_policy: bool = False
) -> sql.Release:
@@ -163,3 +187,226 @@ async def release_validated_and_committee(
if committee is None:
raise RuntimeError(f"Release {project} {version} has no committee")
return release, committee
+
+
+# async def __json_from_maven_cdn(
+# self, api_url: str, group_id: str, artifact_id: str, version: str
+# ) -> outcome.Outcome[models.basic.JSON]:
+# import datetime
+#
+# try:
+# async with util.create_secure_session() as session:
+# async with session.get(api_url) as response:
+# response.raise_for_status()
+#
+# # Use current time as timestamp since we're just validating the package exists
+# timestamp_ms = int(datetime.datetime.now(datetime.UTC).timestamp() * 1000)
+#
+# # Convert to dict matching MavenResponse structure
+# result_dict = {
+# "response": {
+# "start": 0,
+# "docs": [
+# {
+# "g": group_id,
+# "a": artifact_id,
+# "v": version,
+# "timestamp": timestamp_ms,
+# }
+# ],
+# }
+# }
+# result = models.basic.as_json(result_dict)
+# return outcome.Result(result)
+# except aiohttp.ClientError as e:
+# return outcome.Error(e)
+
+
+async def json_from_maven_xml(api_url: str, version: str) -> outcome.Outcome[basic.JSON]:
+ import datetime
+ import xml.etree.ElementTree as ET
+
+ try:
+ async with util.create_secure_session() as session:
+ async with session.get(api_url) as response:
+ response.raise_for_status()
+ xml_text = await response.text()
+
+ # Parse the XML
+ root = ET.fromstring(xml_text)
+
+ # Extract versioning info
+ group = root.find("groupId")
+ artifact = root.find("artifactId")
+ versioning = root.find("versioning")
+ if versioning is None:
+ e = RuntimeError("No versioning element found in Maven metadata")
+ return outcome.Error(e)
+
+ # Get lastUpdated timestamp (format: yyyyMMddHHmmss)
+ last_updated_elem = versioning.find("lastUpdated")
+ if (last_updated_elem is None) or (not last_updated_elem.text):
+ e = RuntimeError("No lastUpdated timestamp found in Maven metadata")
+ return outcome.Error(e)
+
+ # Convert lastUpdated string to Unix timestamp in milliseconds
+ last_updated_str = last_updated_elem.text
+ dt = datetime.datetime.strptime(last_updated_str, "%Y%m%d%H%M%S")
+ dt = dt.replace(tzinfo=datetime.UTC)
+ timestamp_ms = int(dt.timestamp() * 1000)
+
+ # Verify the version exists
+ versions_elem = versioning.find("versions")
+ if versions_elem is not None:
+ versions = [v.text for v in versions_elem.findall("version") if v.text]
+ if version not in versions:
+ e = RuntimeError(f"Version '{version}' not found in Maven metadata")
+ return outcome.Error(e)
+
+ # Convert to dict matching MavenResponse structure
+ result_dict = {
+ "response": {
+ "start": 0,
+ "docs": [
+ {
+ "g": group.text if (group is not None) else "",
+ "a": artifact.text if (artifact is not None) else "",
+ "v": version,
+ "timestamp": timestamp_ms,
+ }
+ ],
+ }
+ }
+ result = basic.as_json(result_dict)
+ return outcome.Result(result)
+ except aiohttp.ClientError as e:
+ return outcome.Error(e)
+ except ET.ParseError as e:
+ return outcome.Error(RuntimeError(f"Failed to parse Maven XML: {e}"))
+
+
+def distribution_upload_date( # noqa: C901
+ platform: sql.DistributionPlatform,
+ data: basic.JSON,
+ version: str,
+) -> datetime.datetime | None:
+ match platform:
+ case sql.DistributionPlatform.ARTIFACT_HUB:
+ if not (versions := distribution.ArtifactHubResponse.model_validate(data).available_versions):
+ return None
+ return datetime.datetime.fromtimestamp(versions[0].ts, tz=datetime.UTC)
+ case sql.DistributionPlatform.DOCKER_HUB:
+ if not (pushed_at := distribution.DockerResponse.model_validate(data).tag_last_pushed):
+ return None
+ return datetime.datetime.fromisoformat(pushed_at.rstrip("Z"))
+ # case models.sql.DistributionPlatform.GITHUB:
+ # if not (published_at := GitHubResponse.model_validate(data).published_at):
+ # return None
+ # return datetime.datetime.fromisoformat(published_at.rstrip("Z"))
+ case sql.DistributionPlatform.MAVEN:
+ m = distribution.MavenResponse.model_validate(data)
+ docs = m.response.docs
+ if not docs:
+ return None
+ timestamp = docs[0].timestamp
+ if not timestamp:
+ return None
+ return datetime.datetime.fromtimestamp(timestamp / 1000, tz=datetime.UTC)
+ case sql.DistributionPlatform.NPM | sql.DistributionPlatform.NPM_SCOPED:
+ if not (times := distribution.NpmResponse.model_validate(data).time):
+ return None
+ # Versions can be in the form "1.2.3" or "v1.2.3", so we check for both
+ if not (upload_time := times.get(version) or times.get(f"v{version}")):
+ return None
+ return datetime.datetime.fromisoformat(upload_time.rstrip("Z"))
+ case sql.DistributionPlatform.PYPI:
+ if not (urls := distribution.PyPIResponse.model_validate(data).urls):
+ return None
+ if not (upload_time := urls[0].upload_time_iso_8601):
+ return None
+ return datetime.datetime.fromisoformat(upload_time.rstrip("Z"))
+ raise NotImplementedError(f"Platform {platform.name} is not yet supported")
+
+
+def distribution_web_url( # noqa: C901
+ platform: sql.DistributionPlatform,
+ data: basic.JSON,
+ version: str,
+) -> str | None:
+ match platform:
+ case sql.DistributionPlatform.ARTIFACT_HUB:
+ ah = distribution.ArtifactHubResponse.model_validate(data)
+ repo_name = ah.repository.name if ah.repository else None
+ pkg_name = ah.name
+ ver = ah.version
+ if repo_name and pkg_name:
+ if ver:
+ return f"https://artifacthub.io/packages/helm/{repo_name}/{pkg_name}/{ver}"
+ return f"https://artifacthub.io/packages/helm/{repo_name}/{pkg_name}/{version}"
+ if ah.home_url:
+ return ah.home_url
+ for link in ah.links:
+ if link.url:
+ return link.url
+ return None
+ case sql.DistributionPlatform.DOCKER_HUB:
+ # The best we can do on Docker Hub is:
+ # f"https://hub.docker.com/_/{package}"
+ return None
+ # case models.sql.DistributionPlatform.GITHUB:
+ # gh = GitHubResponse.model_validate(data)
+ # return gh.html_url
+ case sql.DistributionPlatform.MAVEN:
+ return None
+ case sql.DistributionPlatform.NPM:
+ nr = distribution.NpmResponse.model_validate(data)
+ # return nr.homepage
+ return f"https://www.npmjs.com/package/{nr.name}/v/{version}"
+ case sql.DistributionPlatform.NPM_SCOPED:
+ nr = distribution.NpmResponse.model_validate(data)
+ # TODO: This is not correct
+ return nr.homepage
+ case sql.DistributionPlatform.PYPI:
+ info = distribution.PyPIResponse.model_validate(data).info
+ return info.release_url or info.project_url
+ raise NotImplementedError(f"Platform {platform.name} is not yet supported")
+
+
+def get_api_url(dd: distribution.Data, staging: bool | None = None):
+ template_url = _template_url(dd, staging)
+ api_url = template_url.format(
+ owner_namespace=dd.owner_namespace,
+ package=dd.package,
+ version=dd.version,
+ )
+ if dd.platform == sql.DistributionPlatform.MAVEN:
+ # We do this here because the CDNs break the namespace up into a / delimited URL
+ owner = (dd.owner_namespace or "").replace(".", "/")
+ api_url = template_url.format(
+ owner_namespace=owner,
+ package=dd.package,
+ version=dd.version,
+ )
+ return api_url
+
+
+def _template_url(
+ dd: distribution.Data,
+ staging: bool | None = None,
+) -> str:
+ if staging is False:
+ return dd.platform.value.template_url
+
+ supported = {
+ sql.DistributionPlatform.ARTIFACT_HUB,
+ sql.DistributionPlatform.PYPI,
+ sql.DistributionPlatform.MAVEN,
+ }
+ if dd.platform not in supported:
+ raise RuntimeError("Staging is currently supported only for ArtifactHub, PyPI and Maven Central.")
+
+ template_url = dd.platform.value.template_staging_url
+ if template_url is None:
+ raise RuntimeError("This platform does not provide a staging API endpoint.")
+
+ return template_url
diff --git a/atr/shared/__init__.py b/atr/shared/web.py
similarity index 81%
copy from atr/shared/__init__.py
copy to atr/shared/web.py
index 4c9e400..66c4bc0 100644
--- a/atr/shared/__init__.py
+++ b/atr/shared/web.py
@@ -14,8 +14,7 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
-from typing import TYPE_CHECKING, Final
+from typing import TYPE_CHECKING
import htpy
@@ -27,24 +26,7 @@ import atr.htm as htm
import atr.models.results as results
import atr.models.sql as sql
import atr.post as post
-import atr.shared.announce as announce
-import atr.shared.distribution as distribution
import atr.shared.draft as draft
-import atr.shared.finish as finish
-import atr.shared.ignores as ignores
-import atr.shared.keys as keys
-import atr.shared.manual as manual
-import atr.shared.projects as projects
-import atr.shared.resolve as resolve
-import atr.shared.revisions as revisions
-import atr.shared.sbom as sbom
-import atr.shared.start as start
-import atr.shared.test as test
-import atr.shared.tokens as tokens
-import atr.shared.upload as upload
-import atr.shared.user as user
-import atr.shared.vote as vote
-import atr.shared.voting as voting
import atr.storage as storage
import atr.storage.types as types
import atr.template as template
@@ -55,32 +37,6 @@ if TYPE_CHECKING:
from collections.abc import Sequence
-# | 1 | RSA (Encrypt or Sign) [HAC] |
-# | 2 | RSA Encrypt-Only [HAC] |
-# | 3 | RSA Sign-Only [HAC] |
-# | 16 | Elgamal (Encrypt-Only) [ELGAMAL] [HAC] |
-# | 17 | DSA (Digital Signature Algorithm) [FIPS186] [HAC] |
-# | 18 | ECDH public key algorithm |
-# | 19 | ECDSA public key algorithm [FIPS186] |
-# | 20 | Reserved (formerly Elgamal Encrypt or Sign) |
-# | 21 | Reserved for Diffie-Hellman |
-# | | (X9.42, as defined for IETF-S/MIME) |
-# | 22 | EdDSA [I-D.irtf-cfrg-eddsa] |
-# - https://lists.gnupg.org/pipermail/gnupg-devel/2017-April/032762.html
-
-algorithms: Final[dict[int, str]] = {
- 1: "RSA",
- 2: "RSA",
- 3: "RSA",
- 16: "Elgamal",
- 17: "DSA",
- 18: "ECDH",
- 19: "ECDSA",
- 21: "Diffie-Hellman",
- 22: "EdDSA",
-}
-
-
async def check(
session: web.Committer | None,
release: sql.Release,
@@ -270,27 +226,3 @@ def _warnings_from_vote_result(vote_task: sql.Task | None) -> list[str]:
return ["Vote task result is not a results.VoteInitiate instance."]
return vote_task_result.mail_send_warnings
-
-
-__all__ = [
- "algorithms",
- "announce",
- "check",
- "distribution",
- "draft",
- "finish",
- "ignores",
- "keys",
- "manual",
- "projects",
- "resolve",
- "revisions",
- "sbom",
- "start",
- "test",
- "tokens",
- "upload",
- "user",
- "vote",
- "voting",
-]
diff --git a/atr/storage/writers/distributions.py b/atr/storage/writers/distributions.py
index 74fe074..312a471 100644
--- a/atr/storage/writers/distributions.py
+++ b/atr/storage/writers/distributions.py
@@ -19,16 +19,11 @@
from __future__ import annotations
import datetime
-import sqlite3
-
-import aiohttp
-import sqlalchemy.exc as exc
import atr.db as db
import atr.log as log
-import atr.models.basic as basic
-import atr.models.distribution as distribution
-import atr.models.sql as sql
+import atr.models as models
+import atr.shared.distribution as distribution
import atr.storage as storage
import atr.storage.outcome as outcome
import atr.tasks.gha as gha
@@ -100,7 +95,7 @@ class CommitteeMember(CommitteeParticipant):
async def automate(
self,
release_name: str,
- platform: sql.DistributionPlatform,
+ platform: models.sql.DistributionPlatform,
committee_name: str,
owner_namespace: str | None,
project_name: str,
@@ -110,9 +105,9 @@ class CommitteeMember(CommitteeParticipant):
package: str,
version: str,
staging: bool,
- ) -> sql.Task:
- dist_task = sql.Task(
- task_type=sql.TaskType.DISTRIBUTION_WORKFLOW,
+ ) -> models.sql.Task:
+ dist_task = models.sql.Task(
+ task_type=models.sql.TaskType.DISTRIBUTION_WORKFLOW,
task_args=gha.DistributionWorkflow(
name=release_name,
namespace=owner_namespace or "",
@@ -129,7 +124,7 @@ class CommitteeMember(CommitteeParticipant):
).model_dump(),
asf_uid=util.unwrap(self.__asf_uid),
added=datetime.datetime.now(datetime.UTC),
- status=sql.TaskStatus.QUEUED,
+ status=models.sql.TaskStatus.QUEUED,
project_name=project_name,
version_name=version_name,
revision_number=revision_number,
@@ -142,88 +137,97 @@ class CommitteeMember(CommitteeParticipant):
async def record(
self,
release_name: str,
- platform: sql.DistributionPlatform,
+ platform: models.sql.DistributionPlatform,
owner_namespace: str | None,
package: str,
version: str,
staging: bool,
+ pending: bool,
upload_date: datetime.datetime | None,
- api_url: str,
+ api_url: str | None = None,
web_url: str | None = None,
- ) -> tuple[sql.Distribution, bool]:
- distribution = sql.Distribution(
+ ) -> tuple[models.sql.Distribution, bool]:
+ existing = await self.__data.distribution(release_name, platform, owner_namespace or "", package, version).get()
+ dist = models.sql.Distribution(
platform=platform,
release_name=release_name,
owner_namespace=owner_namespace or "",
package=package,
version=version,
staging=staging,
+ pending=pending,
+ retries=0,
upload_date=upload_date,
api_url=api_url,
web_url=web_url,
)
- self.__data.add(distribution)
- try:
+ if existing is None:
+ self.__data.add(dist)
await self.__data.commit()
- except exc.IntegrityError as e:
- # "The names and numeric values for existing result codes are fixed and unchanging."
- # https://www.sqlite.org/rescode.html
- # e.orig.sqlite_errorcode == 1555
- # e.orig.sqlite_errorname == "SQLITE_CONSTRAINT_PRIMARYKEY"
- match e.orig:
- # TODO: Document this
- case sqlite3.IntegrityError(sqlite_errorcode=sqlite3.SQLITE_CONSTRAINT_PRIMARYKEY):
- if not staging:
- upgraded = await self.__upgrade_staging_to_final(
- release_name,
- platform,
- owner_namespace,
- package,
- version,
- upload_date,
- api_url,
- web_url,
- )
- if upgraded is not None:
- return upgraded, False
- return distribution, False
- raise e
- return distribution, True
+ return dist, True
+ # If we're doing production and existing was for staging, upgrade it
+ if (not staging) and existing.staging:
+ upgraded = await self.__upgrade_staging_to_final(
+ release_name,
+ platform,
+ owner_namespace,
+ package,
+ version,
+ upload_date,
+ api_url,
+ web_url,
+ )
+ if upgraded is not None:
+ return upgraded, False
+ if existing.pending:
+ if pending:
+ existing.retries = existing.retries + 1
+ await self.__data.commit()
+ return existing, False
+ else:
+ existing.pending = False
+ await self.__data.commit()
+ return existing, False
+ return dist, False
async def record_from_data(
self,
release_name: str,
staging: bool,
- dd: distribution.Data,
- ) -> tuple[sql.Distribution, bool, distribution.Metadata]:
- template_url = await self.__template_url(dd, staging)
- api_url = template_url.format(
- owner_namespace=dd.owner_namespace,
- package=dd.package,
- version=dd.version,
- )
- if dd.platform == sql.DistributionPlatform.MAVEN:
- # We do this here because the CDNs break the namespace up into a / delimited URL
- owner = (dd.owner_namespace or "").replace(".", "/")
- api_url = template_url.format(
- owner_namespace=owner,
- package=dd.package,
- version=dd.version,
- )
- api_oc = await self.__json_from_maven_xml(api_url, dd.version)
+ dd: models.distribution.Data,
+ allow_retries: bool = False,
+ ) -> tuple[models.sql.Distribution, bool, models.distribution.Metadata]:
+ api_url = distribution.get_api_url(dd, staging)
+ if dd.platform == models.sql.DistributionPlatform.MAVEN:
+ api_oc = await distribution.json_from_maven_xml(api_url, dd.version)
else:
- api_oc = await self.__json_from_distribution_platform(api_url, dd.platform, dd.version)
+ api_oc = await distribution.json_from_distribution_platform(api_url, dd.platform, dd.version)
match api_oc:
case outcome.Result(result):
pass
case outcome.Error(error):
log.error(f"Failed to get API response from {api_url}: {error}")
+ if allow_retries:
+ dist, added = await self.record(
+ release_name=release_name,
+ platform=dd.platform,
+ owner_namespace=dd.owner_namespace,
+ package=dd.package,
+ version=dd.version,
+ staging=staging,
+ pending=True,
+ upload_date=None,
+ api_url=None,
+ web_url=None,
+ )
+ if added:
+ raise storage.AccessError("Distribution could not be found, ATR will retry this automatically")
raise storage.AccessError(f"Failed to get API response from distribution platform: {error}")
- upload_date = self.__distribution_upload_date(dd.platform, result, dd.version)
+ upload_date = distribution.distribution_upload_date(dd.platform, result, dd.version)
if upload_date is None:
raise storage.AccessError("Failed to get upload date from distribution platform")
- web_url = self.__distribution_web_url(dd.platform, result, dd.version)
- metadata = distribution.Metadata(
+ web_url = distribution.distribution_web_url(dd.platform, result, dd.version)
+ metadata = models.distribution.Metadata(
api_url=api_url,
result=result,
upload_date=upload_date,
@@ -236,244 +240,24 @@ class CommitteeMember(CommitteeParticipant):
package=dd.package,
version=dd.version,
staging=staging,
+ pending=False,
upload_date=upload_date,
api_url=api_url,
web_url=web_url,
)
return dist, added, metadata
- def __distribution_upload_date( # noqa: C901
- self,
- platform: sql.DistributionPlatform,
- data: basic.JSON,
- version: str,
- ) -> datetime.datetime | None:
- match platform:
- case sql.DistributionPlatform.ARTIFACT_HUB:
- if not (versions := distribution.ArtifactHubResponse.model_validate(data).available_versions):
- return None
- return datetime.datetime.fromtimestamp(versions[0].ts, tz=datetime.UTC)
- case sql.DistributionPlatform.DOCKER_HUB:
- if not (pushed_at := distribution.DockerResponse.model_validate(data).tag_last_pushed):
- return None
- return datetime.datetime.fromisoformat(pushed_at.rstrip("Z"))
- # case sql.DistributionPlatform.GITHUB:
- # if not (published_at := GitHubResponse.model_validate(data).published_at):
- # return None
- # return datetime.datetime.fromisoformat(published_at.rstrip("Z"))
- case sql.DistributionPlatform.MAVEN:
- m = distribution.MavenResponse.model_validate(data)
- docs = m.response.docs
- if not docs:
- return None
- timestamp = docs[0].timestamp
- if not timestamp:
- return None
- return datetime.datetime.fromtimestamp(timestamp / 1000, tz=datetime.UTC)
- case sql.DistributionPlatform.NPM | sql.DistributionPlatform.NPM_SCOPED:
- if not (times := distribution.NpmResponse.model_validate(data).time):
- return None
- # Versions can be in the form "1.2.3" or "v1.2.3", so we check for both
- if not (upload_time := times.get(version) or times.get(f"v{version}")):
- return None
- return datetime.datetime.fromisoformat(upload_time.rstrip("Z"))
- case sql.DistributionPlatform.PYPI:
- if not (urls := distribution.PyPIResponse.model_validate(data).urls):
- return None
- if not (upload_time := urls[0].upload_time_iso_8601):
- return None
- return datetime.datetime.fromisoformat(upload_time.rstrip("Z"))
- raise NotImplementedError(f"Platform {platform.name} is not yet supported")
-
- def __distribution_web_url( # noqa: C901
- self,
- platform: sql.DistributionPlatform,
- data: basic.JSON,
- version: str,
- ) -> str | None:
- match platform:
- case sql.DistributionPlatform.ARTIFACT_HUB:
- ah = distribution.ArtifactHubResponse.model_validate(data)
- repo_name = ah.repository.name if ah.repository else None
- pkg_name = ah.name
- ver = ah.version
- if repo_name and pkg_name:
- if ver:
- return f"https://artifacthub.io/packages/helm/{repo_name}/{pkg_name}/{ver}"
- return f"https://artifacthub.io/packages/helm/{repo_name}/{pkg_name}/{version}"
- if ah.home_url:
- return ah.home_url
- for link in ah.links:
- if link.url:
- return link.url
- return None
- case sql.DistributionPlatform.DOCKER_HUB:
- # The best we can do on Docker Hub is:
- # f"https://hub.docker.com/_/{package}"
- return None
- # case sql.DistributionPlatform.GITHUB:
- # gh = GitHubResponse.model_validate(data)
- # return gh.html_url
- case sql.DistributionPlatform.MAVEN:
- return None
- case sql.DistributionPlatform.NPM:
- nr = distribution.NpmResponse.model_validate(data)
- # return nr.homepage
- return f"https://www.npmjs.com/package/{nr.name}/v/{version}"
- case sql.DistributionPlatform.NPM_SCOPED:
- nr = distribution.NpmResponse.model_validate(data)
- # TODO: This is not correct
- return nr.homepage
- case sql.DistributionPlatform.PYPI:
- info = distribution.PyPIResponse.model_validate(data).info
- return info.release_url or info.project_url
- raise NotImplementedError(f"Platform {platform.name} is not yet supported")
-
- async def __json_from_distribution_platform(
- self, api_url: str, platform: sql.DistributionPlatform, version: str
- ) -> outcome.Outcome[basic.JSON]:
- try:
- async with util.create_secure_session() as session:
- async with session.get(api_url) as response:
- response.raise_for_status()
- response_json = await response.json()
- result = basic.as_json(response_json)
- except aiohttp.ClientError as e:
- return outcome.Error(e)
- match platform:
- case sql.DistributionPlatform.NPM | sql.DistributionPlatform.NPM_SCOPED:
- if version not in distribution.NpmResponse.model_validate(result).time:
- e = RuntimeError(f"Version '{version}' not found")
- return outcome.Error(e)
- return outcome.Result(result)
-
- async def __json_from_maven_cdn(
- self, api_url: str, group_id: str, artifact_id: str, version: str
- ) -> outcome.Outcome[basic.JSON]:
- import datetime
-
- try:
- async with util.create_secure_session() as session:
- async with session.get(api_url) as response:
- response.raise_for_status()
-
- # Use current time as timestamp since we're just validating the package exists
- timestamp_ms = int(datetime.datetime.now(datetime.UTC).timestamp() * 1000)
-
- # Convert to dict matching MavenResponse structure
- result_dict = {
- "response": {
- "start": 0,
- "docs": [
- {
- "g": group_id,
- "a": artifact_id,
- "v": version,
- "timestamp": timestamp_ms,
- }
- ],
- }
- }
- result = basic.as_json(result_dict)
- return outcome.Result(result)
- except aiohttp.ClientError as e:
- return outcome.Error(e)
-
- async def __json_from_maven_xml(self, api_url: str, version: str) -> outcome.Outcome[basic.JSON]:
- import datetime
- import xml.etree.ElementTree as ET
-
- try:
- async with util.create_secure_session() as session:
- async with session.get(api_url) as response:
- response.raise_for_status()
- xml_text = await response.text()
-
- # Parse the XML
- root = ET.fromstring(xml_text)
-
- # Extract versioning info
- group = root.find("groupId")
- artifact = root.find("artifactId")
- versioning = root.find("versioning")
- if versioning is None:
- e = RuntimeError("No versioning element found in Maven metadata")
- return outcome.Error(e)
-
- # Get lastUpdated timestamp (format: yyyyMMddHHmmss)
- last_updated_elem = versioning.find("lastUpdated")
- if (last_updated_elem is None) or (not last_updated_elem.text):
- e = RuntimeError("No lastUpdated timestamp found in Maven metadata")
- return outcome.Error(e)
-
- # Convert lastUpdated string to Unix timestamp in milliseconds
- last_updated_str = last_updated_elem.text
- dt = datetime.datetime.strptime(last_updated_str, "%Y%m%d%H%M%S")
- dt = dt.replace(tzinfo=datetime.UTC)
- timestamp_ms = int(dt.timestamp() * 1000)
-
- # Verify the version exists
- versions_elem = versioning.find("versions")
- if versions_elem is not None:
- versions = [v.text for v in versions_elem.findall("version") if v.text]
- if version not in versions:
- e = RuntimeError(f"Version '{version}' not found in Maven metadata")
- return outcome.Error(e)
-
- # Convert to dict matching MavenResponse structure
- result_dict = {
- "response": {
- "start": 0,
- "docs": [
- {
- "g": group.text if (group is not None) else "",
- "a": artifact.text if (artifact is not None) else "",
- "v": version,
- "timestamp": timestamp_ms,
- }
- ],
- }
- }
- result = basic.as_json(result_dict)
- return outcome.Result(result)
- except aiohttp.ClientError as e:
- return outcome.Error(e)
- except ET.ParseError as e:
- return outcome.Error(RuntimeError(f"Failed to parse Maven XML: {e}"))
-
- async def __template_url(
- self,
- dd: distribution.Data,
- staging: bool | None = None,
- ) -> str:
- if staging is False:
- return dd.platform.value.template_url
-
- supported = {
- sql.DistributionPlatform.ARTIFACT_HUB,
- sql.DistributionPlatform.PYPI,
- sql.DistributionPlatform.MAVEN,
- }
- if dd.platform not in supported:
- raise storage.AccessError("Staging is currently supported only for ArtifactHub, PyPI and Maven Central.")
-
- template_url = dd.platform.value.template_staging_url
- if template_url is None:
- raise storage.AccessError("This platform does not provide a staging API endpoint.")
-
- return template_url
-
async def __upgrade_staging_to_final(
self,
release_name: str,
- platform: sql.DistributionPlatform,
+ platform: models.sql.DistributionPlatform,
owner_namespace: str | None,
package: str,
version: str,
upload_date: datetime.datetime | None,
- api_url: str,
+ api_url: str | None,
web_url: str | None,
- ) -> sql.Distribution | None:
+ ) -> models.sql.Distribution | None:
tag = f"{release_name} {platform} {owner_namespace or ''} {package} {version}"
existing = await self.__data.distribution(
release_name=release_name,
@@ -494,7 +278,7 @@ class CommitteeMember(CommitteeParticipant):
async def delete_distribution(
self,
release_name: str,
- platform: sql.DistributionPlatform,
+ platform: models.sql.DistributionPlatform,
owner_namespace: str,
package: str,
version: str,
diff --git a/atr/tasks/__init__.py b/atr/tasks/__init__.py
index 38d57d6..ea613bb 100644
--- a/atr/tasks/__init__.py
+++ b/atr/tasks/__init__.py
@@ -31,6 +31,7 @@ import atr.tasks.checks.rat as rat
import atr.tasks.checks.signature as signature
import atr.tasks.checks.targz as targz
import atr.tasks.checks.zipformat as zipformat
+import atr.tasks.distribution as distribution
import atr.tasks.gha as gha
import atr.tasks.keys as keys
import atr.tasks.message as message
@@ -80,6 +81,33 @@ async def clear_scheduled(caller_data: db.Session | None = None) -> None:
await data.commit()
+async def distribution_status_check(
+ asf_uid: str,
+ caller_data: db.Session | None = None,
+ schedule: datetime.datetime | None = None,
+ schedule_next: bool = False,
+) -> sql.Task:
+ """Queue a workflow status update task."""
+ args = distribution.DistributionStatusCheckArgs(next_schedule_seconds=0, asf_uid=asf_uid)
+ if schedule_next:
+ args.next_schedule_seconds = 2 * 60
+ async with db.ensure_session(caller_data) as data:
+ task = sql.Task(
+ status=sql.TaskStatus.QUEUED,
+ task_type=sql.TaskType.DISTRIBUTION_STATUS,
+ task_args=args.model_dump(),
+ asf_uid=asf_uid,
+ revision_number=None,
+ primary_rel_path=None,
+ )
+ if schedule:
+ task.scheduled = schedule
+ data.add(task)
+ await data.commit()
+ await data.flush()
+ return task
+
+
async def draft_checks(
asf_uid: str, project_name: str, release_version: str, revision_number: str, caller_data: db.Session | None = None
) -> int:
@@ -222,6 +250,8 @@ def queued(
def resolve(task_type: sql.TaskType) -> Callable[..., Awaitable[results.Results | None]]: # noqa: C901
match task_type:
+ case sql.TaskType.DISTRIBUTION_STATUS:
+ return distribution.status_check
case sql.TaskType.DISTRIBUTION_WORKFLOW:
return gha.trigger_workflow
case sql.TaskType.HASHING_CHECK:
diff --git a/atr/tasks/distribution.py b/atr/tasks/distribution.py
new file mode 100644
index 0000000..541f918
--- /dev/null
+++ b/atr/tasks/distribution.py
@@ -0,0 +1,79 @@
+# 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 datetime
+
+import pydantic
+
+import atr.db as db
+import atr.log as log
+import atr.models.distribution as distribution
+import atr.models.results as results
+import atr.models.schema as schema
+import atr.storage as storage
+import atr.tasks as tasks
+import atr.tasks.checks as checks
+
+_RETRY_LIMIT = 5
+
+
+class DistributionStatusCheckArgs(schema.Strict):
+ """Arguments for the task to re-check distribution statuses."""
+
+ next_schedule_seconds: int = pydantic.Field(default=0, description="The next scheduled time")
+ asf_uid: str = schema.description("ASF UID of the user triggering the workflow")
+
+
+@checks.with_model(DistributionStatusCheckArgs)
+async def status_check(args: DistributionStatusCheckArgs, *, task_id: int | None = None) -> results.Results | None:
+ dists = []
+ async with db.session() as data:
+ dists = await data.distribution(pending=True, _with_release=True, _with_release_project=True).all()
+ for dist in dists:
+ dd = distribution.Data(
+ platform=dist.platform,
+ owner_namespace=dist.owner_namespace,
+ package=dist.package,
+ version=dist.version,
+ details=False,
+ )
+ try:
+ async with storage.write_as_committee_member(str(dist.release.project.committee_name), args.asf_uid) as w:
+ if dist.retries >= _RETRY_LIMIT:
+ await w.distributions.delete_distribution(
+ dist.release_name, dist.platform, dist.owner_namespace, dist.package, dist.version
+ )
+ name = f"{dist.platform} {dist.owner_namespace} {dist.package} {dist.version}"
+ log.error(f"Distribution {name} failed {_RETRY_LIMIT} times, skipping")
+ continue
+ await w.distributions.record_from_data(
+ dist.release_name,
+ dist.staging,
+ dd,
+ )
+ except storage.AccessError as e:
+ msg = f"Failed to record distribution: {e}"
+ log.error(msg)
+ raise RuntimeError(msg)
+ if args.next_schedule_seconds:
+ next_schedule = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=args.next_schedule_seconds)
+ await tasks.distribution_status_check(args.asf_uid, schedule=next_schedule, schedule_next=True)
+ log.info(
+ f"Scheduled next workflow status update for: {next_schedule.strftime('%Y-%m-%d %H:%M:%S')}",
+ )
+ return results.DistributionStatusCheck(
+ kind="distribution_status",
+ )
diff --git a/migrations/versions/0042_2026.01.28_3e434625.py b/migrations/versions/0042_2026.01.28_3e434625.py
new file mode 100644
index 0000000..0e96b86
--- /dev/null
+++ b/migrations/versions/0042_2026.01.28_3e434625.py
@@ -0,0 +1,31 @@
+"""columns for pending distributions
+
+Revision ID: 0042_2026.01.28_3e434625
+Revises: 0041_2026.01.22_d1e357f5
+Create Date: 2026-01-28 16:30:09.232235+00:00
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# Revision identifiers, used by Alembic
+revision: str = "0042_2026.01.28_3e434625"
+down_revision: str | None = "0041_2026.01.22_d1e357f5"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+ with op.batch_alter_table("distribution", schema=None) as batch_op:
+ batch_op.add_column(sa.Column("pending", sa.Boolean(), nullable=False, server_default=sa.false()))
+ batch_op.add_column(sa.Column("retries", sa.Integer(), nullable=False, server_default="0"))
+ batch_op.alter_column("api_url", existing_type=sa.VARCHAR(), nullable=True)
+
+
+def downgrade() -> None:
+ with op.batch_alter_table("distribution", schema=None) as batch_op:
+ batch_op.alter_column("api_url", existing_type=sa.VARCHAR(), nullable=False)
+ batch_op.drop_column("retries")
+ batch_op.drop_column("pending")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org