Verified Commit a77d44e9 authored by Kevin Morris's avatar Kevin Morris
Browse files

change(python): move comaint routes to pkgbase router



Also brings over comaint utility functions to the pkgbase
package.

Signed-off-by: Kevin Morris's avatarKevin Morris <kevr@0cost.org>
parent bd2ad9b6
Pipeline #14250 waiting for manual action with stages
in 3 minutes and 12 seconds
......@@ -4,17 +4,15 @@ from typing import Dict, List, Tuple, Union
import orjson
from fastapi import HTTPException, Request
from fastapi import HTTPException
from sqlalchemy import orm
from aurweb import config, db, l10n, models, util
from aurweb.models import Package, PackageBase, User
from aurweb import config, db, models
from aurweb.models import Package
from aurweb.models.official_provider import OFFICIAL_BASE, OfficialProvider
from aurweb.models.package_comaintainer import PackageComaintainer
from aurweb.models.package_dependency import PackageDependency
from aurweb.models.package_relation import PackageRelation
from aurweb.redis import redis_connection
from aurweb.scripts import notify
from aurweb.templates import register_filter
Providers = List[Union[PackageRelation, OfficialProvider]]
......@@ -222,151 +220,6 @@ def query_notified(query: List[models.Package],
return output
def remove_comaintainer(comaint: PackageComaintainer) \
-> notify.ComaintainerRemoveNotification:
"""
Remove a PackageComaintainer.
This function does *not* begin any database transaction and
must be used **within** a database transaction, e.g.:
with db.begin():
remove_comaintainer(comaint)
:param comaint: Target PackageComaintainer to be deleted
:return: ComaintainerRemoveNotification
"""
pkgbase = comaint.PackageBase
notif = notify.ComaintainerRemoveNotification(comaint.User.ID, pkgbase.ID)
db.delete(comaint)
rotate_comaintainers(pkgbase)
return notif
def remove_comaintainers(pkgbase: PackageBase, usernames: List[str]) -> None:
"""
Remove comaintainers from `pkgbase`.
:param pkgbase: PackageBase instance
:param usernames: Iterable of username strings
"""
notifications = []
with db.begin():
comaintainers = pkgbase.comaintainers.join(User).filter(
User.Username.in_(usernames)
).all()
notifications = [
notify.ComaintainerRemoveNotification(co.User.ID, pkgbase.ID)
for co in comaintainers
]
db.delete_all(comaintainers)
# Rotate comaintainer priority values.
with db.begin():
rotate_comaintainers(pkgbase)
# Send out notifications.
util.apply_all(notifications, lambda n: n.send())
def latest_priority(pkgbase: PackageBase) -> int:
"""
Return the highest Priority column related to `pkgbase`.
:param pkgbase: PackageBase instance
:return: Highest Priority found or 0 if no records exist
"""
# Order comaintainers related to pkgbase by Priority DESC.
record = pkgbase.comaintainers.order_by(
PackageComaintainer.Priority.desc()).first()
# Use Priority column if record exists, otherwise 0.
return record.Priority if record else 0
class NoopComaintainerNotification:
""" A noop notification stub used as an error-state return value. """
def send(self) -> None:
""" noop """
return
def add_comaintainer(pkgbase: PackageBase, comaintainer: User) \
-> notify.ComaintainerAddNotification:
"""
Add a new comaintainer to `pkgbase`.
:param pkgbase: PackageBase instance
:param comaintainer: User instance used for new comaintainer record
:return: ComaintainerAddNotification
"""
# Skip given `comaintainers` who are already maintainer.
if pkgbase.Maintainer == comaintainer:
return NoopComaintainerNotification()
# Priority for the new comaintainer is +1 more than the highest.
new_prio = latest_priority(pkgbase) + 1
with db.begin():
db.create(PackageComaintainer, PackageBase=pkgbase,
User=comaintainer, Priority=new_prio)
return notify.ComaintainerAddNotification(comaintainer.ID, pkgbase.ID)
def add_comaintainers(request: Request, pkgbase: models.PackageBase,
usernames: List[str]) -> None:
"""
Add comaintainers to `pkgbase`.
:param request: FastAPI request
:param pkgbase: PackageBase instance
:param usernames: Iterable of username strings
:return: Error string on failure else None
"""
# For each username in usernames, perform validation of the username
# and append the User record to `users` if no errors occur.
users = []
for username in usernames:
user = db.query(User).filter(User.Username == username).first()
if not user:
_ = l10n.get_translator_for_request(request)
return _("Invalid user name: %s") % username
users.append(user)
notifications = []
def add_comaint(user: User):
nonlocal notifications
# Populate `notifications` with add_comaintainer's return value,
# which is a ComaintainerAddNotification.
notifications.append(add_comaintainer(pkgbase, user))
# Move along: add all `users` as new `pkgbase` comaintainers.
util.apply_all(users, add_comaint)
# Send out notifications.
util.apply_all(notifications, lambda n: n.send())
def rotate_comaintainers(pkgbase: PackageBase) -> None:
"""
Rotate `pkgbase` comaintainers.
This function resets the Priority column of all PackageComaintainer
instances related to `pkgbase` to seqential 1 .. n values with
persisted order.
:param pkgbase: PackageBase instance
"""
comaintainers = pkgbase.comaintainers.order_by(
models.PackageComaintainer.Priority.asc())
for i, comaint in enumerate(comaintainers):
comaint.Priority = i + 1
def pkg_required(pkgname: str, provides: List[str], limit: int) \
-> List[PackageDependency]:
"""
......
......@@ -9,6 +9,7 @@ from aurweb.models.package_comaintainer import PackageComaintainer
from aurweb.models.package_notification import PackageNotification
from aurweb.models.request_type import DELETION_ID, MERGE_ID, ORPHAN_ID
from aurweb.packages.requests import handle_request, update_closure_comment
from aurweb.pkgbase import util as pkgbaseutil
from aurweb.scripts import notify, popupdate
logger = logging.get_logger(__name__)
......@@ -47,8 +48,6 @@ def pkgbase_unflag_instance(request: Request, pkgbase: PackageBase) -> None:
def pkgbase_disown_instance(request: Request, pkgbase: PackageBase) -> None:
import aurweb.packages.util as pkgutil
disowner = request.user
notifs = [notify.DisownNotification(disowner.ID, pkgbase.ID)]
......@@ -62,7 +61,7 @@ def pkgbase_disown_instance(request: Request, pkgbase: PackageBase) -> None:
if prio_comaint:
# If there is such a comaintainer, promote them to maint.
pkgbase.Maintainer = prio_comaint.User
notifs.append(pkgutil.remove_comaintainer(prio_comaint))
notifs.append(pkgbaseutil.remove_comaintainer(prio_comaint))
else:
# Otherwise, just orphan the package completely.
pkgbase.Maintainer = None
......
from typing import Any, Dict
from typing import Any, Dict, List
from fastapi import Request
from aurweb import config
from aurweb.models import PackageBase
from aurweb import config, db, l10n, util
from aurweb.models import PackageBase, User
from aurweb.models.package_comaintainer import PackageComaintainer
from aurweb.models.package_comment import PackageComment
from aurweb.models.package_request import PackageRequest
from aurweb.models.package_vote import PackageVote
from aurweb.scripts import notify
from aurweb.templates import make_context as _make_context
......@@ -45,3 +47,148 @@ def make_context(request: Request, pkgbase: PackageBase) -> Dict[str, Any]:
).count()
return context
def remove_comaintainer(comaint: PackageComaintainer) \
-> notify.ComaintainerRemoveNotification:
"""
Remove a PackageComaintainer.
This function does *not* begin any database transaction and
must be used **within** a database transaction, e.g.:
with db.begin():
remove_comaintainer(comaint)
:param comaint: Target PackageComaintainer to be deleted
:return: ComaintainerRemoveNotification
"""
pkgbase = comaint.PackageBase
notif = notify.ComaintainerRemoveNotification(comaint.User.ID, pkgbase.ID)
db.delete(comaint)
rotate_comaintainers(pkgbase)
return notif
def remove_comaintainers(pkgbase: PackageBase, usernames: List[str]) -> None:
"""
Remove comaintainers from `pkgbase`.
:param pkgbase: PackageBase instance
:param usernames: Iterable of username strings
"""
notifications = []
with db.begin():
comaintainers = pkgbase.comaintainers.join(User).filter(
User.Username.in_(usernames)
).all()
notifications = [
notify.ComaintainerRemoveNotification(co.User.ID, pkgbase.ID)
for co in comaintainers
]
db.delete_all(comaintainers)
# Rotate comaintainer priority values.
with db.begin():
rotate_comaintainers(pkgbase)
# Send out notifications.
util.apply_all(notifications, lambda n: n.send())
def latest_priority(pkgbase: PackageBase) -> int:
"""
Return the highest Priority column related to `pkgbase`.
:param pkgbase: PackageBase instance
:return: Highest Priority found or 0 if no records exist
"""
# Order comaintainers related to pkgbase by Priority DESC.
record = pkgbase.comaintainers.order_by(
PackageComaintainer.Priority.desc()).first()
# Use Priority column if record exists, otherwise 0.
return record.Priority if record else 0
class NoopComaintainerNotification:
""" A noop notification stub used as an error-state return value. """
def send(self) -> None:
""" noop """
return
def add_comaintainer(pkgbase: PackageBase, comaintainer: User) \
-> notify.ComaintainerAddNotification:
"""
Add a new comaintainer to `pkgbase`.
:param pkgbase: PackageBase instance
:param comaintainer: User instance used for new comaintainer record
:return: ComaintainerAddNotification
"""
# Skip given `comaintainers` who are already maintainer.
if pkgbase.Maintainer == comaintainer:
return NoopComaintainerNotification()
# Priority for the new comaintainer is +1 more than the highest.
new_prio = latest_priority(pkgbase) + 1
with db.begin():
db.create(PackageComaintainer, PackageBase=pkgbase,
User=comaintainer, Priority=new_prio)
return notify.ComaintainerAddNotification(comaintainer.ID, pkgbase.ID)
def add_comaintainers(request: Request, pkgbase: PackageBase,
usernames: List[str]) -> None:
"""
Add comaintainers to `pkgbase`.
:param request: FastAPI request
:param pkgbase: PackageBase instance
:param usernames: Iterable of username strings
:return: Error string on failure else None
"""
# For each username in usernames, perform validation of the username
# and append the User record to `users` if no errors occur.
users = []
for username in usernames:
user = db.query(User).filter(User.Username == username).first()
if not user:
_ = l10n.get_translator_for_request(request)
return _("Invalid user name: %s") % username
users.append(user)
notifications = []
def add_comaint(user: User):
nonlocal notifications
# Populate `notifications` with add_comaintainer's return value,
# which is a ComaintainerAddNotification.
notifications.append(add_comaintainer(pkgbase, user))
# Move along: add all `users` as new `pkgbase` comaintainers.
util.apply_all(users, add_comaint)
# Send out notifications.
util.apply_all(notifications, lambda n: n.send())
def rotate_comaintainers(pkgbase: PackageBase) -> None:
"""
Rotate `pkgbase` comaintainers.
This function resets the Priority column of all PackageComaintainer
instances related to `pkgbase` to seqential 1 .. n values with
persisted order.
:param pkgbase: PackageBase instance
"""
comaintainers = pkgbase.comaintainers.order_by(
PackageComaintainer.Priority.asc())
for i, comaint in enumerate(comaintainers):
comaint.Priority = i + 1
......@@ -181,73 +181,6 @@ async def package(request: Request, name: str) -> Response:
return render_template(request, "packages/show.html", context)
@router.get("/pkgbase/{name}/comaintainers")
@auth_required()
async def package_base_comaintainers(request: Request, name: str) -> Response:
# Get the PackageBase.
pkgbase = get_pkg_or_base(name, models.PackageBase)
# Unauthorized users (Non-TU/Dev and not the pkgbase maintainer)
# get redirected to the package base's page.
has_creds = request.user.has_credential(creds.PKGBASE_EDIT_COMAINTAINERS,
approved=[pkgbase.Maintainer])
if not has_creds:
return RedirectResponse(f"/pkgbase/{name}",
status_code=HTTPStatus.SEE_OTHER)
# Add our base information.
context = make_context(request, "Manage Co-maintainers")
context["pkgbase"] = pkgbase
context["comaintainers"] = [
c.User.Username for c in pkgbase.comaintainers
]
return render_template(request, "pkgbase/comaintainers.html", context)
@router.post("/pkgbase/{name}/comaintainers")
@auth_required()
async def package_base_comaintainers_post(
request: Request, name: str,
users: str = Form(default=str())) -> Response:
# Get the PackageBase.
pkgbase = get_pkg_or_base(name, models.PackageBase)
# Unauthorized users (Non-TU/Dev and not the pkgbase maintainer)
# get redirected to the package base's page.
has_creds = request.user.has_credential(creds.PKGBASE_EDIT_COMAINTAINERS,
approved=[pkgbase.Maintainer])
if not has_creds:
return RedirectResponse(f"/pkgbase/{name}",
status_code=HTTPStatus.SEE_OTHER)
users = {e.strip() for e in users.split("\n") if bool(e.strip())}
records = {c.User.Username for c in pkgbase.comaintainers}
users_to_rm = records.difference(users)
pkgutil.remove_comaintainers(pkgbase, users_to_rm)
logger.debug(f"{request.user} removed comaintainers from "
f"{pkgbase.Name}: {users_to_rm}")
users_to_add = users.difference(records)
error = pkgutil.add_comaintainers(request, pkgbase, users_to_add)
if error:
context = make_context(request, "Manage Co-maintainers")
context["pkgbase"] = pkgbase
context["comaintainers"] = [
c.User.Username for c in pkgbase.comaintainers
]
context["errors"] = [error]
return render_template(request, "pkgbase/comaintainers.html", context)
logger.debug(f"{request.user} added comaintainers to "
f"{pkgbase.Name}: {users_to_add}")
return RedirectResponse(f"/pkgbase/{pkgbase.Name}",
status_code=HTTPStatus.SEE_OTHER)
@router.get("/requests")
@auth_required()
async def requests(request: Request,
......
......@@ -5,7 +5,7 @@ from fastapi import APIRouter, Form, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse, RedirectResponse
from sqlalchemy import and_
from aurweb import db, l10n, templates, util
from aurweb import db, l10n, logging, templates, util
from aurweb.auth import auth_required, creds
from aurweb.exceptions import InvariantError
from aurweb.models import PackageBase
......@@ -23,6 +23,7 @@ from aurweb.scripts import popupdate
from aurweb.scripts.rendercomment import update_comment_render_fastapi
from aurweb.templates import make_variable_context, render_template
logger = logging.get_logger(__name__)
router = APIRouter()
......@@ -572,6 +573,74 @@ async def pkgbase_adopt_post(request: Request, name: str):
status_code=HTTPStatus.SEE_OTHER)
@router.get("/pkgbase/{name}/comaintainers")
@auth_required()
async def pkgbase_comaintainers(request: Request, name: str) -> Response:
# Get the PackageBase.
pkgbase = get_pkg_or_base(name, PackageBase)
# Unauthorized users (Non-TU/Dev and not the pkgbase maintainer)
# get redirected to the package base's page.
has_creds = request.user.has_credential(creds.PKGBASE_EDIT_COMAINTAINERS,
approved=[pkgbase.Maintainer])
if not has_creds:
return RedirectResponse(f"/pkgbase/{name}",
status_code=HTTPStatus.SEE_OTHER)
# Add our base information.
context = templates.make_context(request, "Manage Co-maintainers")
context.update({
"pkgbase": pkgbase,
"comaintainers": [
c.User.Username for c in pkgbase.comaintainers
]
})
return render_template(request, "pkgbase/comaintainers.html", context)
@router.post("/pkgbase/{name}/comaintainers")
@auth_required()
async def pkgbase_comaintainers_post(request: Request, name: str,
users: str = Form(default=str())) \
-> Response:
# Get the PackageBase.
pkgbase = get_pkg_or_base(name, PackageBase)
# Unauthorized users (Non-TU/Dev and not the pkgbase maintainer)
# get redirected to the package base's page.
has_creds = request.user.has_credential(creds.PKGBASE_EDIT_COMAINTAINERS,
approved=[pkgbase.Maintainer])
if not has_creds:
return RedirectResponse(f"/pkgbase/{name}",
status_code=HTTPStatus.SEE_OTHER)
users = {e.strip() for e in users.split("\n") if bool(e.strip())}
records = {c.User.Username for c in pkgbase.comaintainers}
users_to_rm = records.difference(users)
pkgbaseutil.remove_comaintainers(pkgbase, users_to_rm)
logger.debug(f"{request.user} removed comaintainers from "
f"{pkgbase.Name}: {users_to_rm}")
users_to_add = users.difference(records)
error = pkgbaseutil.add_comaintainers(request, pkgbase, users_to_add)
if error:
context = templates.make_context(request, "Manage Co-maintainers")
context["pkgbase"] = pkgbase
context["comaintainers"] = [
c.User.Username for c in pkgbase.comaintainers
]
context["errors"] = [error]
return render_template(request, "pkgbase/comaintainers.html", context)
logger.debug(f"{request.user} added comaintainers to "
f"{pkgbase.Name}: {users_to_add}")
return RedirectResponse(f"/pkgbase/{pkgbase.Name}",
status_code=HTTPStatus.SEE_OTHER)
@router.get("/pkgbase/{name}/delete")
@auth_required()
async def pkgbase_delete_get(request: Request, name: str):
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment