Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ dependencies = [
"greenlet>=3.0.0",
"authlib>=1.4.0",
"itsdangerous>=2.2.0",
"python-multipart>=0.0.20",
]

[project.optional-dependencies]
Expand Down
223 changes: 165 additions & 58 deletions src/squishmark/routers/admin.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,17 @@
"""Admin routes for notes, analytics, and cache management."""

import json
import logging
from typing import Annotated, Any

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from pydantic import BaseModel, ValidationError
from sqlalchemy.ext.asyncio import AsyncSession

from squishmark.config import get_settings
from squishmark.models.content import Config
from squishmark.models.db import get_db_session
from squishmark.models.db import Note, get_db_session
from squishmark.services.analytics import AnalyticsService
from squishmark.services.cache import get_cache
from squishmark.services.github import get_github_service
Expand All @@ -22,6 +23,11 @@
router = APIRouter(prefix="/admin", tags=["admin"])


def is_htmx(request: Request) -> bool:
"""Return True when the request was made by HTMX."""
return request.headers.get("HX-Request") == "true"


# Pydantic models for request/response
class NoteCreate(BaseModel):
"""Request body for creating a note."""
Expand Down Expand Up @@ -66,6 +72,9 @@ async def get_current_admin(request: Request) -> str:

Raises HTTPException 401 if not authenticated.
Raises HTTPException 403 if not an admin.

For HTMX requests, attaches an ``HX-Redirect`` header so the browser
is redirected to the login page without any client JavaScript.
"""
settings = get_settings()

Expand All @@ -74,14 +83,16 @@ async def get_current_admin(request: Request) -> str:
logger.warning("Auth bypassed - returning dev-admin user")
return "dev-admin"

htmx_headers = {"HX-Redirect": "/auth/login"} if is_htmx(request) else None

# Check for user in session (set by OAuth callback)
user = request.session.get("user") if hasattr(request, "session") else None

if user is None:
raise HTTPException(status_code=401, detail="Not authenticated")
raise HTTPException(status_code=401, detail="Not authenticated", headers=htmx_headers)

if user["login"] not in settings.admin_users_list:
raise HTTPException(status_code=403, detail="Not authorized")
raise HTTPException(status_code=403, detail="Not authorized", headers=htmx_headers)

return user["login"]

Expand All @@ -90,6 +101,93 @@ async def get_current_admin(request: Request) -> str:
DbSession = Annotated[AsyncSession, Depends(get_db_session)]


def _to_note_response(note: Note) -> NoteResponse:
"""Convert an ORM ``Note`` to the JSON-serializable ``NoteResponse``."""
return NoteResponse(
id=note.id,
path=note.path,
text=note.text,
is_public=note.is_public,
author=note.author,
created_at=note.created_at.isoformat(),
updated_at=note.updated_at.isoformat(),
)


async def _render_note_partial(template_name: str, **context: Any) -> str:
"""Render a notes admin partial using the active site theme."""
github_service = get_github_service()
config_data = await github_service.get_config()
theme_name = Config.from_dict(config_data).theme.name
theme_engine = await get_theme_engine(github_service)
return theme_engine.render_partial(template_name, theme_override=theme_name, **context)


def _is_form_request(request: Request) -> bool:
content_type = request.headers.get("content-type", "")
return content_type.startswith(("application/x-www-form-urlencoded", "multipart/form-data"))


async def _load_json_body(request: Request) -> dict:
"""Decode a JSON body, raising 422 on malformed JSON to match FastAPI defaults."""
try:
return await request.json()
except json.JSONDecodeError as exc:
raise HTTPException(status_code=422, detail=f"Invalid JSON: {exc.msg}") from exc


async def parse_note_create(request: Request) -> NoteCreate:
"""Parse a ``NoteCreate`` payload from either form data or JSON.

HTMX submits standard HTML forms as ``application/x-www-form-urlencoded``.
Non-HTMX API callers continue to send JSON. Validation errors are returned
as 422 (matching FastAPI's default body-binding behavior) regardless of
which path produced them.
"""
if _is_form_request(request):
form = await request.form()
# Pass through what was actually sent: absent fields stay absent so
# Pydantic enforces required-ness instead of silently coercing to "".
data: dict[str, Any] = {"is_public": "is_public" in form}
if "path" in form:
data["path"] = str(form["path"])
if "text" in form:
data["text"] = str(form["text"])
else:
data = await _load_json_body(request)
try:
return NoteCreate.model_validate(data)
except ValidationError as exc:
raise HTTPException(status_code=422, detail=exc.errors()) from exc


async def parse_note_update(request: Request) -> NoteUpdate:
"""Parse a ``NoteUpdate`` payload from either form data or JSON.

Form semantics:
- ``is_public`` is always set explicitly (True if the checkbox is present,
False otherwise) — never None — so unchecking the checkbox actually
persists ``is_public=False`` rather than being treated as "no change".
- ``text`` is left absent (``None``, meaning "no change") when the form
doesn't send it. The HTMX edit form always submits ``text`` because the
field is ``required``, so this only affects programmatic form callers
that intentionally omit the field.

Validation errors are returned as 422 from both paths.
"""
if _is_form_request(request):
form = await request.form()
data: dict[str, Any] = {"is_public": "is_public" in form}
if "text" in form:
data["text"] = str(form["text"])
else:
data = await _load_json_body(request)
try:
return NoteUpdate.model_validate(data)
except ValidationError as exc:
raise HTTPException(status_code=422, detail=exc.errors()) from exc


# Admin dashboard
@router.get("", response_class=HTMLResponse)
async def admin_dashboard(
Expand Down Expand Up @@ -122,18 +220,7 @@ async def admin_dashboard(
config,
user={"login": admin},
analytics=analytics,
notes=[
NoteResponse(
id=n.id,
path=n.path,
text=n.text,
is_public=n.is_public,
author=n.author,
created_at=n.created_at.isoformat(),
updated_at=n.updated_at.isoformat(),
)
for n in notes
],
notes=[_to_note_response(n) for n in notes],
cache_size=cache.size,
)
except Exception:
Expand Down Expand Up @@ -180,55 +267,44 @@ async def list_notes(
session: DbSession,
) -> list[NoteResponse]:
"""List all notes."""
del admin # auth side-effect only
notes_service = NotesService(session)
notes = await notes_service.get_all()
return [
NoteResponse(
id=n.id,
path=n.path,
text=n.text,
is_public=n.is_public,
author=n.author,
created_at=n.created_at.isoformat(),
updated_at=n.updated_at.isoformat(),
)
for n in notes
]
return [_to_note_response(n) for n in notes]


@router.post("/notes", status_code=201)
@router.post("/notes", status_code=201, response_model=None)
async def create_note(
request: Request,
admin: AdminUser,
session: DbSession,
note_data: NoteCreate,
) -> NoteResponse:
"""Create a new note."""
) -> NoteResponse | HTMLResponse:
"""Create a new note. Returns an HTML partial for HTMX, JSON otherwise."""
note_data = await parse_note_create(request)
notes_service = NotesService(session)
note = await notes_service.create(
path=note_data.path,
text=note_data.text,
author=admin,
is_public=note_data.is_public,
)
return NoteResponse(
id=note.id,
path=note.path,
text=note.text,
is_public=note.is_public,
author=note.author,
created_at=note.created_at.isoformat(),
updated_at=note.updated_at.isoformat(),
)
response = _to_note_response(note)
if is_htmx(request):
html = await _render_note_partial("admin/_note_item.html", note=response)
return HTMLResponse(content=html, status_code=201)
return response


@router.put("/notes/{note_id}")
@router.put("/notes/{note_id}", response_model=None)
async def update_note(
request: Request,
admin: AdminUser,
session: DbSession,
note_id: int,
note_data: NoteUpdate,
) -> NoteResponse:
"""Update a note."""
) -> NoteResponse | HTMLResponse:
"""Update a note. Returns an HTML partial for HTMX, JSON otherwise."""
del admin # auth side-effect only
note_data = await parse_note_update(request)
notes_service = NotesService(session)
note = await notes_service.update_note(
note_id=note_id,
Expand All @@ -237,32 +313,63 @@ async def update_note(
)
if note is None:
raise HTTPException(status_code=404, detail="Note not found")

return NoteResponse(
id=note.id,
path=note.path,
text=note.text,
is_public=note.is_public,
author=note.author,
created_at=note.created_at.isoformat(),
updated_at=note.updated_at.isoformat(),
)
response = _to_note_response(note)
if is_htmx(request):
html = await _render_note_partial("admin/_note_item.html", note=response)
return HTMLResponse(content=html)
return response


@router.delete("/notes/{note_id}")
@router.delete("/notes/{note_id}", response_model=None)
async def delete_note(
request: Request,
admin: AdminUser,
session: DbSession,
note_id: int,
) -> dict[str, str]:
"""Delete a note."""
) -> dict[str, str] | HTMLResponse:
"""Delete a note. Returns an empty 200 for HTMX so the row is removed."""
del admin # auth side-effect only
notes_service = NotesService(session)
deleted = await notes_service.delete(note_id)
if not deleted:
raise HTTPException(status_code=404, detail="Note not found")
if is_htmx(request):
return HTMLResponse(content="", status_code=200)
return {"status": "deleted"}


@router.get("/notes/{note_id}/edit", response_class=HTMLResponse)
async def edit_note_form(
admin: AdminUser,
session: DbSession,
note_id: int,
) -> HTMLResponse:
"""Return the inline edit form for a note (HTMX swap target)."""
del admin # auth side-effect only
notes_service = NotesService(session)
note = await notes_service.get_by_id(note_id)
if note is None:
raise HTTPException(status_code=404, detail="Note not found")
html = await _render_note_partial("admin/_note_edit_form.html", note=_to_note_response(note))
return HTMLResponse(content=html)


@router.get("/notes/{note_id}/view", response_class=HTMLResponse)
async def view_note(
admin: AdminUser,
session: DbSession,
note_id: int,
) -> HTMLResponse:
"""Return the read-only note row (used by the edit form's Cancel button)."""
del admin # auth side-effect only
notes_service = NotesService(session)
note = await notes_service.get_by_id(note_id)
if note is None:
raise HTTPException(status_code=404, detail="Note not found")
html = await _render_note_partial("admin/_note_item.html", note=_to_note_response(note))
return HTMLResponse(content=html)


# Cache management
@router.post("/cache/refresh")
async def refresh_cache(
Expand Down
24 changes: 24 additions & 0 deletions src/squishmark/services/theme/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,30 @@ async def render_admin(
"""Render the admin dashboard."""
return await self.render("admin/admin.html", config, theme_override=theme_override, **context)

def render_partial(
self,
template_name: str,
theme_override: str | None = None,
**context: Any,
) -> str:
"""Render a small HTML fragment without the heavy default context.

For HTMX swaps where we only need a self-contained snippet (no nav,
no favicon, no GitHub fetches). The partial must not depend on
site/theme/canonical context.

The previous ``current_theme`` is restored after rendering so concurrent
requests don't see each other's theme leak through this shared loader.
"""
previous_theme = self.loader.current_theme
try:
if theme_override:
self.loader.current_theme = theme_override
template = self.env.get_template(template_name)
return template.render(**context)
finally:
self.loader.current_theme = previous_theme


# Global theme engine instance
_theme_engine: ThemeEngine | None = None
Expand Down
Loading
Loading