Make template rendering compatible across Starlette versions

This commit is contained in:
Krikorios
2026-05-07 13:05:24 +03:00
parent aed2fdc801
commit 68d13dbf57
6 changed files with 71 additions and 37 deletions
+34 -21
View File
@@ -1,6 +1,7 @@
from fastapi import APIRouter, Depends, Request, Form, HTTPException, status from fastapi import APIRouter, Depends, Request, Form, HTTPException, status
from fastapi.responses import HTMLResponse, RedirectResponse, FileResponse from fastapi.responses import HTMLResponse, RedirectResponse, FileResponse
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from services.template_utils import render_template
from services.auth_service import ( from services.auth_service import (
verify_password, get_user_by_username, create_user, delete_user, get_all_users, verify_password, get_user_by_username, create_user, delete_user, get_all_users,
create_session, get_session, delete_session, SESSION_TTL_SECONDS, create_session, get_session, delete_session, SESSION_TTL_SECONDS,
@@ -45,7 +46,7 @@ def require_admin(request: Request):
@router.get("/login", response_class=HTMLResponse) @router.get("/login", response_class=HTMLResponse)
async def login_get(request: Request): async def login_get(request: Request):
return templates.TemplateResponse("login.html", {"request": request, "error": None}) return render_template(templates, request, "login.html", {"error": None})
@router.post("/login", response_class=HTMLResponse) @router.post("/login", response_class=HTMLResponse)
async def login_post(request: Request, username: str = Form(...), password: str = Form(...)): async def login_post(request: Request, username: str = Form(...), password: str = Form(...)):
@@ -53,18 +54,22 @@ async def login_post(request: Request, username: str = Form(...), password: str
ip = _client_ip(request) ip = _client_ip(request)
if is_login_blocked(username, ip): if is_login_blocked(username, ip):
return templates.TemplateResponse( return render_template(
templates,
request,
"login.html", "login.html",
{"request": request, "error": "تم حجب محاولات تسجيل الدخول مؤقتاً. حاول بعد 15 دقيقة."}, {"error": "تم حجب محاولات تسجيل الدخول مؤقتاً. حاول بعد 15 دقيقة."},
status_code=429, status_code=429,
) )
user = get_user_by_username(username) user = get_user_by_username(username)
if not user or not verify_password(user["password_hash"], password): if not user or not verify_password(user["password_hash"], password):
record_login_attempt(username, ip, success=False) record_login_attempt(username, ip, success=False)
return templates.TemplateResponse( return render_template(
templates,
request,
"login.html", "login.html",
{"request": request, "error": "Invalid username or password"}, {"error": "Invalid username or password"},
status_code=401, status_code=401,
) )
@@ -106,10 +111,7 @@ async def logout(request: Request):
@router.get("/users", response_class=HTMLResponse) @router.get("/users", response_class=HTMLResponse)
async def users_list(request: Request, current=Depends(require_admin)): async def users_list(request: Request, current=Depends(require_admin)):
users = get_all_users() users = get_all_users()
return templates.TemplateResponse( return render_template(templates, request, "users.html", {"users": users, "current_user": current})
"users.html",
{"request": request, "users": users, "current_user": current},
)
@router.post("/users/create") @router.post("/users/create")
async def add_user( async def add_user(
@@ -124,15 +126,19 @@ async def add_user(
role = "user" role = "user"
if len(password) < 8: if len(password) < 8:
users = get_all_users() users = get_all_users()
return templates.TemplateResponse( return render_template(
templates,
request,
"users.html", "users.html",
{"request": request, "users": users, "current_user": current, "error": "Password must be at least 8 characters."}, {"users": users, "current_user": current, "error": "Password must be at least 8 characters."},
) )
if get_user_by_username(username): if get_user_by_username(username):
users = get_all_users() users = get_all_users()
return templates.TemplateResponse( return render_template(
templates,
request,
"users.html", "users.html",
{"request": request, "users": users, "current_user": current, "error": f"User '{username}' already exists."}, {"users": users, "current_user": current, "error": f"User '{username}' already exists."},
) )
create_user(username, password, role=role) create_user(username, password, role=role)
return RedirectResponse(url="/auth/users", status_code=status.HTTP_303_SEE_OTHER) return RedirectResponse(url="/auth/users", status_code=status.HTTP_303_SEE_OTHER)
@@ -141,18 +147,22 @@ async def add_user(
async def remove_user(request: Request, user_id: int, current=Depends(require_admin)): async def remove_user(request: Request, user_id: int, current=Depends(require_admin)):
if user_id == current["user_id"]: if user_id == current["user_id"]:
users = get_all_users() users = get_all_users()
return templates.TemplateResponse( return render_template(
templates,
request,
"users.html", "users.html",
{"request": request, "users": users, "current_user": current, "error": "لا يمكنك حذف حسابك الخاص."}, {"users": users, "current_user": current, "error": "لا يمكنك حذف حسابك الخاص."},
status_code=400, status_code=400,
) )
# Prevent removing the last admin # Prevent removing the last admin
target = next((u for u in get_all_users() if u["id"] == user_id), None) target = next((u for u in get_all_users() if u["id"] == user_id), None)
if target and target["role"] == "admin" and count_admins() <= 1: if target and target["role"] == "admin" and count_admins() <= 1:
users = get_all_users() users = get_all_users()
return templates.TemplateResponse( return render_template(
templates,
request,
"users.html", "users.html",
{"request": request, "users": users, "current_user": current, "error": "لا يمكن حذف آخر مسؤول في النظام."}, {"users": users, "current_user": current, "error": "لا يمكن حذف آخر مسؤول في النظام."},
status_code=400, status_code=400,
) )
delete_user(user_id) delete_user(user_id)
@@ -172,9 +182,11 @@ async def change_role(
target = next((u for u in get_all_users() if u["id"] == user_id), None) target = next((u for u in get_all_users() if u["id"] == user_id), None)
if target and target["role"] == "admin" and role != "admin" and count_admins() <= 1: if target and target["role"] == "admin" and role != "admin" and count_admins() <= 1:
users = get_all_users() users = get_all_users()
return templates.TemplateResponse( return render_template(
templates,
request,
"users.html", "users.html",
{"request": request, "users": users, "current_user": current, "error": "لا يمكن تخفيض رتبة آخر مسؤول."}, {"users": users, "current_user": current, "error": "لا يمكن تخفيض رتبة آخر مسؤول."},
status_code=400, status_code=400,
) )
set_user_role(user_id, role) set_user_role(user_id, role)
@@ -187,10 +199,11 @@ async def backup_db(request: Request, current=Depends(require_admin)):
return FileResponse(backup_path, media_type="application/octet-stream", filename=os.path.basename(backup_path)) return FileResponse(backup_path, media_type="application/octet-stream", filename=os.path.basename(backup_path))
except Exception as e: except Exception as e:
msg = f"Backup failed: {str(e)}" msg = f"Backup failed: {str(e)}"
return templates.TemplateResponse( return render_template(
templates,
request,
"users.html", "users.html",
{ {
"request": request,
"users": get_all_users(), "users": get_all_users(),
"current_user": current, "current_user": current,
"backup_msg": msg, "backup_msg": msg,
+8 -4
View File
@@ -7,6 +7,7 @@ from fastapi.templating import Jinja2Templates
from config import UPLOAD_DIR from config import UPLOAD_DIR
from database.connection import get_db from database.connection import get_db
from services.template_utils import render_template
from services.thumbnails import get_or_create_thumbnail from services.thumbnails import get_or_create_thumbnail
router = APIRouter() router = APIRouter()
@@ -103,10 +104,11 @@ async def document_queue(
total_pages = max(1, (total_filtered + PAGE_SIZE - 1) // PAGE_SIZE) total_pages = max(1, (total_filtered + PAGE_SIZE - 1) // PAGE_SIZE)
return templates.TemplateResponse( return render_template(
templates,
request,
"documents.html", "documents.html",
{ {
"request": request,
"documents": [dict(r) for r in rows], "documents": [dict(r) for r in rows],
"stats": dict(stats) if stats else {}, "stats": dict(stats) if stats else {},
"current_status": status, "current_status": status,
@@ -368,9 +370,11 @@ async def duplicates_view(request: Request):
continue continue
groups.append({"keeper": keeper, "duplicates": dups_by_keeper.get(k, [])}) groups.append({"keeper": keeper, "duplicates": dups_by_keeper.get(k, [])})
return templates.TemplateResponse( return render_template(
templates,
request,
"duplicates.html", "duplicates.html",
{"request": request, "groups": groups, "total_duplicates": len(dup_rows)}, {"groups": groups, "total_duplicates": len(dup_rows)},
) )
+4 -2
View File
@@ -16,6 +16,7 @@ from services.extractor import (
verify_page_correlation, verify_page_correlation,
) )
from services.search_service import normalize_arabic, _normalize_scope from services.search_service import normalize_arabic, _normalize_scope
from services.template_utils import render_template
router = APIRouter() router = APIRouter()
templates = Jinja2Templates(directory="templates") templates = Jinja2Templates(directory="templates")
@@ -328,10 +329,11 @@ async def review_document(request: Request, doc_id: int, wait: int = 0):
</body></html>""" </body></html>"""
) )
return templates.TemplateResponse( return render_template(
templates,
request,
"review.html", "review.html",
{ {
"request": request,
"doc": doc, "doc": doc,
"upload_dir": "/uploads", "upload_dir": "/uploads",
"providers": get_available_providers(), "providers": get_available_providers(),
+8 -8
View File
@@ -1,6 +1,7 @@
from fastapi import APIRouter, Request from fastapi import APIRouter, Request
from fastapi.responses import Response from fastapi.responses import Response
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from services.template_utils import render_template
from services.search_service import ( from services.search_service import (
get_person_with_properties, get_person_with_properties,
@@ -37,10 +38,11 @@ async def search(
persons[0].get("search_scope"), persons[0].get("search_scope"),
) )
return templates.TemplateResponse( return render_template(
templates,
request,
"search.html", "search.html",
{ {
"request": request,
"q": q, "q": q,
"property_number": property_number, "property_number": property_number,
"district": district, "district": district,
@@ -56,10 +58,11 @@ async def search(
async def person_detail(request: Request, person_id: int, search_scope: str = ""): async def person_detail(request: Request, person_id: int, search_scope: str = ""):
data = get_person_with_properties(person_id, search_scope.strip() or None) data = get_person_with_properties(person_id, search_scope.strip() or None)
if not data: if not data:
return templates.TemplateResponse( return render_template(
templates,
request,
"search.html", "search.html",
{ {
"request": request,
"error": "Person not found", "error": "Person not found",
"q": "", "q": "",
"persons": [], "persons": [],
@@ -71,10 +74,7 @@ async def person_detail(request: Request, person_id: int, search_scope: str = ""
}, },
status_code=404, status_code=404,
) )
return templates.TemplateResponse( return render_template(templates, request, "person_detail.html", data)
"person_detail.html",
{"request": request, **data},
)
@router.get("/persons/{person_id}/export") @router.get("/persons/{person_id}/export")
+4 -2
View File
@@ -13,6 +13,7 @@ from config import UPLOAD_DIR, MAX_CONCURRENT_EXTRACTIONS
from database.connection import get_db from database.connection import get_db
from services.extractor import extract_document, get_available_providers, get_default_provider from services.extractor import extract_document, get_available_providers, get_default_provider
from services.pdf_handler import pdf_to_images from services.pdf_handler import pdf_to_images
from services.template_utils import render_template
router = APIRouter() router = APIRouter()
templates = Jinja2Templates(directory="templates") templates = Jinja2Templates(directory="templates")
@@ -204,10 +205,11 @@ async def upload_page(request: Request):
SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) AS errors SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) AS errors
FROM documents WHERE status != 'staged'""" FROM documents WHERE status != 'staged'"""
).fetchone() ).fetchone()
return templates.TemplateResponse( return render_template(
templates,
request,
"index.html", "index.html",
{ {
"request": request,
"stats": dict(stats) if stats else {}, "stats": dict(stats) if stats else {},
"providers": get_available_providers(), "providers": get_available_providers(),
"default_provider": get_default_provider(), "default_provider": get_default_provider(),
+13
View File
@@ -0,0 +1,13 @@
import inspect
from collections.abc import Mapping
def render_template(templates, request, template_name: str, context: Mapping | None = None, **response_kwargs):
context_dict = dict(context or {})
parameters = list(inspect.signature(templates.TemplateResponse).parameters)
if parameters and parameters[0] == "request":
return templates.TemplateResponse(request, template_name, context_dict, **response_kwargs)
context_dict.setdefault("request", request)
return templates.TemplateResponse(template_name, context_dict, **response_kwargs)