Files
rose-ash/blog/bp/blog/admin/routes.py
giles 293f7713d6
Some checks failed
Build and Deploy / build-and-deploy (push) Failing after 16s
Auto-mount defpages: eliminate Python route stubs across all 9 services
Defpages are now declared with absolute paths in .sx files and auto-mounted
directly on the Quart app, removing ~850 lines of blueprint mount_pages calls,
before_request hooks, and g.* wrapper boilerplate. A new page = one defpage
declaration, nothing else.

Infrastructure:
- async_eval awaits coroutine results from callable dispatch
- auto_mount_pages() mounts all registered defpages on the app
- g._defpage_ctx pattern passes helper data to layout context

Migrated: sx, account, orders, federation, cart, market, events, blog

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-03 19:03:15 +00:00

118 lines
3.3 KiB
Python

from __future__ import annotations
import re
from quart import (
Blueprint,
redirect,
url_for,
request,
g,
)
from sqlalchemy import select, delete
from shared.browser.app.authz import require_admin
from shared.browser.app.redis_cacher import invalidate_tag_cache
from models.tag_group import TagGroup, TagGroupTag
from models.ghost_content import Tag
def _slugify(name: str) -> str:
s = name.strip().lower()
s = re.sub(r"[^\w\s-]", "", s)
s = re.sub(r"[\s_]+", "-", s)
return s.strip("-")
async def _unassigned_tags(session):
"""Return public, non-deleted tags not assigned to any group."""
assigned_sq = select(TagGroupTag.tag_id).subquery()
q = (
select(Tag)
.where(
Tag.deleted_at.is_(None),
(Tag.visibility == "public") | (Tag.visibility.is_(None)),
Tag.id.notin_(select(assigned_sq)),
)
.order_by(Tag.name)
)
return list((await session.execute(q)).scalars())
def register():
bp = Blueprint("tag_groups_admin", __name__, url_prefix="/settings/tag-groups")
@bp.post("/")
@require_admin
async def create():
form = await request.form
name = (form.get("name") or "").strip()
if not name:
return redirect(url_for("defpage_tag_groups_page"))
slug = _slugify(name)
feature_image = (form.get("feature_image") or "").strip() or None
colour = (form.get("colour") or "").strip() or None
sort_order = int(form.get("sort_order") or 0)
tg = TagGroup(
name=name, slug=slug,
feature_image=feature_image, colour=colour,
sort_order=sort_order,
)
g.s.add(tg)
await g.s.flush()
await invalidate_tag_cache("blog")
return redirect(url_for("defpage_tag_groups_page"))
@bp.post("/<int:id>/")
@require_admin
async def save(id: int):
tg = await g.s.get(TagGroup, id)
if not tg:
return redirect(url_for("defpage_tag_groups_page"))
form = await request.form
name = (form.get("name") or "").strip()
if name:
tg.name = name
tg.slug = _slugify(name)
tg.feature_image = (form.get("feature_image") or "").strip() or None
tg.colour = (form.get("colour") or "").strip() or None
tg.sort_order = int(form.get("sort_order") or 0)
# Update tag assignments
selected_tag_ids = set()
for val in form.getlist("tag_ids"):
try:
selected_tag_ids.add(int(val))
except (ValueError, TypeError):
pass
# Remove old assignments
await g.s.execute(
delete(TagGroupTag).where(TagGroupTag.tag_group_id == id)
)
await g.s.flush()
# Add new assignments
for tid in selected_tag_ids:
g.s.add(TagGroupTag(tag_group_id=id, tag_id=tid))
await g.s.flush()
await invalidate_tag_cache("blog")
return redirect(url_for("defpage_tag_group_edit", id=id))
@bp.post("/<int:id>/delete/")
@require_admin
async def delete_group(id: int):
tg = await g.s.get(TagGroup, id)
if tg:
await g.s.delete(tg)
await g.s.flush()
await invalidate_tag_cache("blog")
return redirect(url_for("defpage_tag_groups_page"))
return bp