122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
"""Admin Trash: view, restore, and permanently delete soft-deleted items."""
|
|
|
|
from fastapi import APIRouter, Request, Depends, Form
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import text
|
|
from typing import Optional
|
|
|
|
from core.database import get_db
|
|
from core.base_repository import BaseRepository
|
|
from core.sidebar import get_sidebar_data
|
|
|
|
router = APIRouter(prefix="/admin/trash", tags=["admin"])
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
# Entity configs for trash view: (table, display_name, name_column, detail_url_pattern)
|
|
TRASH_ENTITIES = [
|
|
{"table": "tasks", "label": "Tasks", "name_col": "title", "url": "/tasks/{id}"},
|
|
{"table": "projects", "label": "Projects", "name_col": "name", "url": "/projects/{id}"},
|
|
{"table": "notes", "label": "Notes", "name_col": "title", "url": "/notes/{id}"},
|
|
{"table": "links", "label": "Links", "name_col": "label", "url": "/links"},
|
|
{"table": "contacts", "label": "Contacts", "name_col": "first_name", "url": "/contacts/{id}"},
|
|
{"table": "domains", "label": "Domains", "name_col": "name", "url": "/domains"},
|
|
{"table": "areas", "label": "Areas", "name_col": "name", "url": "/areas"},
|
|
{"table": "lists", "label": "Lists", "name_col": "name", "url": "/lists/{id}"},
|
|
{"table": "capture", "label": "Capture", "name_col": "raw_text", "url": "/capture"},
|
|
{"table": "daily_focus", "label": "Focus Items", "name_col": "id", "url": "/focus"},
|
|
]
|
|
|
|
|
|
@router.get("/")
|
|
async def trash_view(
|
|
request: Request,
|
|
entity_type: Optional[str] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
sidebar = await get_sidebar_data(db)
|
|
deleted_items = []
|
|
entity_counts = {}
|
|
|
|
for entity in TRASH_ENTITIES:
|
|
try:
|
|
# Count deleted items per type
|
|
result = await db.execute(text(
|
|
f"SELECT count(*) FROM {entity['table']} WHERE is_deleted = true"
|
|
))
|
|
count = result.scalar() or 0
|
|
entity_counts[entity["table"]] = count
|
|
|
|
# Load items for selected type (or all if none selected)
|
|
if count > 0 and (entity_type is None or entity_type == entity["table"]):
|
|
result = await db.execute(text(f"""
|
|
SELECT id, {entity['name_col']} as display_name, deleted_at, created_at
|
|
FROM {entity['table']}
|
|
WHERE is_deleted = true
|
|
ORDER BY deleted_at DESC
|
|
LIMIT 50
|
|
"""))
|
|
rows = [dict(r._mapping) for r in result]
|
|
for row in rows:
|
|
deleted_items.append({
|
|
"id": str(row["id"]),
|
|
"name": str(row.get("display_name") or row["id"])[:100],
|
|
"table": entity["table"],
|
|
"type_label": entity["label"],
|
|
"deleted_at": row.get("deleted_at"),
|
|
"created_at": row.get("created_at"),
|
|
})
|
|
except Exception:
|
|
continue
|
|
|
|
total_deleted = sum(entity_counts.values())
|
|
|
|
return templates.TemplateResponse("trash.html", {
|
|
"request": request, "sidebar": sidebar,
|
|
"deleted_items": deleted_items,
|
|
"entity_counts": entity_counts,
|
|
"trash_entities": TRASH_ENTITIES,
|
|
"current_type": entity_type or "",
|
|
"total_deleted": total_deleted,
|
|
"page_title": "Trash", "active_nav": "trash",
|
|
})
|
|
|
|
|
|
@router.post("/{table}/{item_id}/restore")
|
|
async def restore_item(table: str, item_id: str, request: Request, db: AsyncSession = Depends(get_db)):
|
|
# Validate table name against known entities
|
|
valid_tables = {e["table"] for e in TRASH_ENTITIES}
|
|
if table not in valid_tables:
|
|
return RedirectResponse(url="/admin/trash", status_code=303)
|
|
|
|
repo = BaseRepository(table, db)
|
|
await repo.restore(item_id)
|
|
referer = request.headers.get("referer", "/admin/trash")
|
|
return RedirectResponse(url=referer, status_code=303)
|
|
|
|
|
|
@router.post("/{table}/{item_id}/permanent-delete")
|
|
async def permanent_delete_item(table: str, item_id: str, request: Request, db: AsyncSession = Depends(get_db)):
|
|
valid_tables = {e["table"] for e in TRASH_ENTITIES}
|
|
if table not in valid_tables:
|
|
return RedirectResponse(url="/admin/trash", status_code=303)
|
|
|
|
repo = BaseRepository(table, db)
|
|
await repo.permanent_delete(item_id)
|
|
referer = request.headers.get("referer", "/admin/trash")
|
|
return RedirectResponse(url=referer, status_code=303)
|
|
|
|
|
|
@router.post("/empty")
|
|
async def empty_trash(request: Request, db: AsyncSession = Depends(get_db)):
|
|
"""Permanently delete ALL soft-deleted items across all tables."""
|
|
for entity in TRASH_ENTITIES:
|
|
try:
|
|
await db.execute(text(
|
|
f"DELETE FROM {entity['table']} WHERE is_deleted = true"
|
|
))
|
|
except Exception:
|
|
continue
|
|
return RedirectResponse(url="/admin/trash", status_code=303)
|