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-releases.git
The following commit(s) were added to refs/heads/main by this push:
new 8177448 Make the forms to add, delete, and update ignores more type
safe
8177448 is described below
commit 8177448bde990094a64e78585325a58fd3145759
Author: Sean B. Palmer <[email protected]>
AuthorDate: Wed Nov 12 20:26:56 2025 +0000
Make the forms to add, delete, and update ignores more type safe
---
atr/get/ignores.py | 83 ++++++++++++++++------------
atr/post/ignores.py | 102 +++++++++++++++-------------------
atr/shared/ignores.py | 149 +++++++++++++++++++++++++++++++++++++-------------
3 files changed, 202 insertions(+), 132 deletions(-)
diff --git a/atr/get/ignores.py b/atr/get/ignores.py
index e32e604..91c29bf 100644
--- a/atr/get/ignores.py
+++ b/atr/get/ignores.py
@@ -18,10 +18,9 @@
from typing import Final
import markupsafe
-import wtforms
import atr.blueprints.get as get
-import atr.forms as forms
+import atr.form as form
import atr.htm as htm
import atr.models.sql as sql
import atr.post as post
@@ -66,55 +65,69 @@ async def ignores(session: web.Committer, committee_name:
str) -> str | web.Werk
return await template.blank("Ignored checks", content)
+def _add_ignore(committee_name: str) -> htm.Element:
+ form_path = util.as_url(post.ignores.ignores,
committee_name=committee_name)
+ block = htm.Block(htm.div)
+ block.h2["Add ignore"]
+ block.p["Add a new ignore for a check result."]
+ form.render_block(
+ block,
+ model_cls=shared.ignores.AddIgnoreForm,
+ action=form_path,
+ submit_label="Add ignore",
+ )
+ return block.collect()
+
+
def _check_result_ignore_card(cri: sql.CheckResultIgnore) -> htm.Element:
h3_id = cri.id or ""
h3_asf_uid = cri.asf_uid
h3_created = util.format_datetime(cri.created)
card_header_h3 = htm.h3(".mt-3.mb-0")[f"{h3_id} - {h3_asf_uid} -
{h3_created}"]
- form_update = shared.ignores.UpdateIgnoreForm(id=cri.id)
-
- def set_field(field: wtforms.StringField | wtforms.SelectField, value: str
| None) -> None:
- if value is not None:
- field.data = value
-
- set_field(form_update.release_glob, cri.release_glob)
- set_field(form_update.revision_number, cri.revision_number)
- set_field(form_update.checker_glob, cri.checker_glob)
- set_field(form_update.primary_rel_path_glob, cri.primary_rel_path_glob)
- set_field(form_update.member_rel_path_glob, cri.member_rel_path_glob)
- set_field(form_update.status, cri.status.to_form_field() if cri.status
else "None")
- set_field(form_update.message_glob, cri.message_glob)
-
- form_path_update = util.as_url(post.ignores.ignores_committee_update,
committee_name=cri.committee_name)
- form_update_html = forms.render_table(form_update, form_path_update)
+ # Update form
+ update_form_block = htm.Block(htm.div)
+ form_path_update = util.as_url(post.ignores.ignores,
committee_name=cri.committee_name)
+ status = shared.ignores.sql_to_ignore_status(cri.status)
+ form.render_block(
+ update_form_block,
+ model_cls=shared.ignores.UpdateIgnoreForm,
+ action=form_path_update,
+ submit_label="Update ignore",
+ form_classes="",
+ defaults={
+ "id": cri.id or 0,
+ "release_glob": cri.release_glob or "",
+ "revision_number": cri.revision_number or "",
+ "checker_glob": cri.checker_glob or "",
+ "primary_rel_path_glob": cri.primary_rel_path_glob or "",
+ "member_rel_path_glob": cri.member_rel_path_glob or "",
+ "status": status,
+ "message_glob": cri.message_glob or "",
+ },
+ )
- form_delete = shared.ignores.DeleteIgnoreForm(id=cri.id)
- form_path_delete = util.as_url(post.ignores.ignores_committee_delete,
committee_name=cri.committee_name)
- form_delete_html = forms.render_simple(
- form_delete,
- form_path_delete,
- form_classes=".mt-2.mb-0",
+ # Delete form
+ delete_form_block = htm.Block(htm.div)
+ form.render_block(
+ delete_form_block,
+ model_cls=shared.ignores.DeleteIgnoreForm,
+ action=form_path_update,
+ submit_label="Delete",
submit_classes="btn-danger",
+ form_classes=".mt-2.mb-0",
+ defaults={"id": cri.id or 0},
+ empty=True,
)
card = htm.div(".card.mb-5")[
- htm.div(".card-header.d-flex.justify-content-between")[card_header_h3,
form_delete_html],
- htm.div(".card-body")[form_update_html],
+ htm.div(".card-header.d-flex.justify-content-between")[card_header_h3,
delete_form_block.collect()],
+ htm.div(".card-body")[update_form_block.collect()],
]
return card
-def _add_ignore(committee_name: str) -> htm.Element:
- form_path = util.as_url(post.ignores.ignores_committee_add,
committee_name=committee_name)
- return htm.div[
- htm.h2["Add ignore"],
- htm.p["Add a new ignore for a check result."],
- forms.render_columns(shared.ignores.AddIgnoreForm(), form_path),
- ]
-
-
def _existing_ignores(ignores: list[sql.CheckResultIgnore]) -> htm.Element:
return htm.div[
htm.h2["Existing ignores"],
diff --git a/atr/post/ignores.py b/atr/post/ignores.py
index 988a396..0fb38fe 100644
--- a/atr/post/ignores.py
+++ b/atr/post/ignores.py
@@ -16,35 +16,46 @@
# under the License.
-import quart
-
import atr.blueprints.post as post
import atr.get as get
-import atr.models.sql as sql
import atr.shared as shared
import atr.storage as storage
import atr.web as web
[email protected]("/ignores/<committee_name>/add")
-async def ignores_committee_add(session: web.Committer, committee_name: str)
-> str | web.WerkzeugResponse:
- data = await quart.request.form
- form = await shared.ignores.AddIgnoreForm.create_form(data=data)
- if not (await form.validate_on_submit()):
- return await session.redirect(get.ignores.ignores, error="Form
validation errors")
[email protected]("/ignores/<committee_name>")
[email protected](shared.ignores.IgnoreForm)
+async def ignores(
+ session: web.Committer, ignore_form: shared.ignores.IgnoreForm,
committee_name: str
+) -> web.WerkzeugResponse:
+ """Handle forms on the ignores page."""
+ match ignore_form:
+ case shared.ignores.AddIgnoreForm() as add_form:
+ return await _add_ignore(session, add_form, committee_name)
+
+ case shared.ignores.DeleteIgnoreForm() as delete_form:
+ return await _delete_ignore(session, delete_form, committee_name)
+
+ case shared.ignores.UpdateIgnoreForm() as update_form:
+ return await _update_ignore(session, update_form, committee_name)
- status = sql.CheckResultStatusIgnore.from_form_field(form.status.data)
+
+async def _add_ignore(
+ session: web.Committer, add_form: shared.ignores.AddIgnoreForm,
committee_name: str
+) -> web.WerkzeugResponse:
+ """Add a new ignore."""
+ status = shared.ignores.ignore_status_to_sql(add_form.status) # type:
ignore[arg-type]
async with storage.write() as write:
wacm = write.as_committee_member(committee_name)
await wacm.checks.ignore_add(
- release_glob=form.release_glob.data or None,
- revision_number=form.revision_number.data or None,
- checker_glob=form.checker_glob.data or None,
- primary_rel_path_glob=form.primary_rel_path_glob.data or None,
- member_rel_path_glob=form.member_rel_path_glob.data or None,
+ release_glob=add_form.release_glob or None,
+ revision_number=add_form.revision_number or None,
+ checker_glob=add_form.checker_glob or None,
+ primary_rel_path_glob=add_form.primary_rel_path_glob or None,
+ member_rel_path_glob=add_form.member_rel_path_glob or None,
status=status,
- message_glob=form.message_glob.data or None,
+ message_glob=add_form.message_glob or None,
)
return await session.redirect(
@@ -54,28 +65,13 @@ async def ignores_committee_add(session: web.Committer,
committee_name: str) ->
)
[email protected]("/ignores/<committee_name>/delete")
-async def ignores_committee_delete(session: web.Committer, committee_name:
str) -> str | web.WerkzeugResponse:
- data = await quart.request.form
- form = await shared.ignores.DeleteIgnoreForm.create_form(data=data)
- if not (await form.validate_on_submit()):
- return await session.redirect(
- get.ignores.ignores,
- committee_name=committee_name,
- error="Form validation errors",
- )
-
- if not isinstance(form.id.data, str):
- return await session.redirect(
- get.ignores.ignores,
- committee_name=committee_name,
- error="Invalid ignore ID",
- )
-
- cri_id = int(form.id.data)
+async def _delete_ignore(
+ session: web.Committer, delete_form: shared.ignores.DeleteIgnoreForm,
committee_name: str
+) -> web.WerkzeugResponse:
+ """Delete an ignore."""
async with storage.write() as write:
wacm = write.as_committee_member(committee_name)
- await wacm.checks.ignore_delete(id=cri_id)
+ await wacm.checks.ignore_delete(id=delete_form.id)
return await session.redirect(
get.ignores.ignores,
@@ -84,33 +80,23 @@ async def ignores_committee_delete(session: web.Committer,
committee_name: str)
)
[email protected]("/ignores/<committee_name>/update")
-async def ignores_committee_update(session: web.Committer, committee_name:
str) -> str | web.WerkzeugResponse:
- data = await quart.request.form
- form = await shared.ignores.UpdateIgnoreForm.create_form(data=data)
- if not (await form.validate_on_submit()):
- return await session.redirect(get.ignores.ignores, error="Form
validation errors")
-
- status = sql.CheckResultStatusIgnore.from_form_field(form.status.data)
- if not isinstance(form.id.data, str):
- return await session.redirect(
- get.ignores.ignores,
- committee_name=committee_name,
- error="Invalid ignore ID",
- )
- cri_id = int(form.id.data)
+async def _update_ignore(
+ session: web.Committer, update_form: shared.ignores.UpdateIgnoreForm,
committee_name: str
+) -> web.WerkzeugResponse:
+ """Update an ignore."""
+ status = shared.ignores.ignore_status_to_sql(update_form.status) # type:
ignore[arg-type]
async with storage.write() as write:
wacm = write.as_committee_member(committee_name)
await wacm.checks.ignore_update(
- id=cri_id,
- release_glob=form.release_glob.data or None,
- revision_number=form.revision_number.data or None,
- checker_glob=form.checker_glob.data or None,
- primary_rel_path_glob=form.primary_rel_path_glob.data or None,
- member_rel_path_glob=form.member_rel_path_glob.data or None,
+ id=update_form.id,
+ release_glob=update_form.release_glob or None,
+ revision_number=update_form.revision_number or None,
+ checker_glob=update_form.checker_glob or None,
+ primary_rel_path_glob=update_form.primary_rel_path_glob or None,
+ member_rel_path_glob=update_form.member_rel_path_glob or None,
status=status,
- message_glob=form.message_glob.data or None,
+ message_glob=update_form.message_glob or None,
)
return await session.redirect(
diff --git a/atr/shared/ignores.py b/atr/shared/ignores.py
index c06632e..79c41cc 100644
--- a/atr/shared/ignores.py
+++ b/atr/shared/ignores.py
@@ -15,54 +15,125 @@
# specific language governing permissions and limitations
# under the License.
+"""ignores.py"""
-import atr.forms as forms
+import enum
+from typing import Annotated, Literal
+
+import pydantic
+
+import atr.form as form
import atr.models.sql as sql
+type ADD = Literal["add"]
+type DELETE = Literal["delete"]
+type UPDATE = Literal["update"]
+
+
+class IgnoreStatus(enum.Enum):
+ """Wrapper enum for ignore status."""
+
+ NO_STATUS = "-"
+ EXCEPTION = "Exception"
+ FAILURE = "Failure"
+ WARNING = "Warning"
+
+
+def ignore_status_to_sql(status: IgnoreStatus | None) ->
sql.CheckResultStatusIgnore | None:
+ """Convert wrapper enum to SQL enum."""
+ if (status is None) or (status == IgnoreStatus.NO_STATUS):
+ return None
+ match status:
+ case IgnoreStatus.EXCEPTION:
+ return sql.CheckResultStatusIgnore.EXCEPTION
+ case IgnoreStatus.FAILURE:
+ return sql.CheckResultStatusIgnore.FAILURE
+ case IgnoreStatus.WARNING:
+ return sql.CheckResultStatusIgnore.WARNING
-class AddIgnoreForm(forms.Typed):
- # TODO: Validate that at least one field is set
- release_glob = forms.optional("Release pattern")
- revision_number = forms.optional("Revision number (literal)")
- checker_glob = forms.optional("Checker pattern")
- primary_rel_path_glob = forms.optional("Primary rel path pattern")
- member_rel_path_glob = forms.optional("Member rel path pattern")
- status = forms.select(
+
+def sql_to_ignore_status(status: sql.CheckResultStatusIgnore | None) ->
IgnoreStatus:
+ """Convert SQL enum to wrapper enum."""
+ if status is None:
+ return IgnoreStatus.NO_STATUS
+ match status:
+ case sql.CheckResultStatusIgnore.EXCEPTION:
+ return IgnoreStatus.EXCEPTION
+ case sql.CheckResultStatusIgnore.FAILURE:
+ return IgnoreStatus.FAILURE
+ case sql.CheckResultStatusIgnore.WARNING:
+ return IgnoreStatus.WARNING
+
+
+class AddIgnoreForm(form.Form):
+ variant: ADD = form.value(ADD)
+ release_glob: str = form.label("Release pattern", default="")
+ revision_number: str = form.label("Revision number (literal)", default="")
+ checker_glob: str = form.label("Checker pattern", default="")
+ primary_rel_path_glob: str = form.label("Primary rel path pattern",
default="")
+ member_rel_path_glob: str = form.label("Member rel path pattern",
default="")
+ status: form.Enum[IgnoreStatus] = form.label(
"Status",
- optional=True,
- choices=[
- (None, "-"),
- (sql.CheckResultStatusIgnore.EXCEPTION, "Exception"),
- (sql.CheckResultStatusIgnore.FAILURE, "Failure"),
- (sql.CheckResultStatusIgnore.WARNING, "Warning"),
- ],
+ widget=form.Widget.SELECT,
)
- message_glob = forms.optional("Message pattern")
- submit = forms.submit("Add ignore")
+ message_glob: str = form.label("Message pattern", default="")
+
+ @pydantic.model_validator(mode="after")
+ def validate_at_least_one_field(self) -> "AddIgnoreForm":
+ has_status = self.status != IgnoreStatus.NO_STATUS # type:
ignore[comparison-overlap]
+ if not any(
+ [
+ self.release_glob,
+ self.revision_number,
+ self.checker_glob,
+ self.primary_rel_path_glob,
+ self.member_rel_path_glob,
+ has_status,
+ self.message_glob,
+ ]
+ ):
+ raise ValueError("At least one field must be set")
+ return self
-class DeleteIgnoreForm(forms.Typed):
- id = forms.hidden()
- submit = forms.submit("Delete")
+class DeleteIgnoreForm(form.Form):
+ variant: DELETE = form.value(DELETE)
+ id: int = form.label("ID", widget=form.Widget.HIDDEN)
-class UpdateIgnoreForm(forms.Typed):
- # TODO: Validate that at least one field is set
- id = forms.hidden()
- release_glob = forms.optional("Release pattern")
- revision_number = forms.optional("Revision number (literal)")
- checker_glob = forms.optional("Checker pattern")
- primary_rel_path_glob = forms.optional("Primary rel path pattern")
- member_rel_path_glob = forms.optional("Member rel path pattern")
- status = forms.select(
+class UpdateIgnoreForm(form.Form):
+ variant: UPDATE = form.value(UPDATE)
+ id: int = form.label("ID", widget=form.Widget.HIDDEN)
+ release_glob: str = form.label("Release pattern", default="")
+ revision_number: str = form.label("Revision number (literal)", default="")
+ checker_glob: str = form.label("Checker pattern", default="")
+ primary_rel_path_glob: str = form.label("Primary rel path pattern",
default="")
+ member_rel_path_glob: str = form.label("Member rel path pattern",
default="")
+ status: form.Enum[IgnoreStatus] = form.label(
"Status",
- optional=True,
- choices=[
- (None, "-"),
- (sql.CheckResultStatusIgnore.EXCEPTION, "Exception"),
- (sql.CheckResultStatusIgnore.FAILURE, "Failure"),
- (sql.CheckResultStatusIgnore.WARNING, "Warning"),
- ],
+ widget=form.Widget.SELECT,
)
- message_glob = forms.optional("Message pattern")
- submit = forms.submit("Update ignore")
+ message_glob: str = form.label("Message pattern", default="")
+
+ @pydantic.model_validator(mode="after")
+ def validate_at_least_one_field(self) -> "UpdateIgnoreForm":
+ has_status = self.status != IgnoreStatus.NO_STATUS # type:
ignore[comparison-overlap]
+ if not any(
+ [
+ self.release_glob,
+ self.revision_number,
+ self.checker_glob,
+ self.primary_rel_path_glob,
+ self.member_rel_path_glob,
+ has_status,
+ self.message_glob,
+ ]
+ ):
+ raise ValueError("At least one field must be set")
+ return self
+
+
+type IgnoreForm = Annotated[
+ AddIgnoreForm | DeleteIgnoreForm | UpdateIgnoreForm,
+ form.DISCRIMINATOR,
+]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
