Coverage for src / lexigram / admin / controllers / draft_controller.py: 0%

1from __future__ import annotations 

2 

3from datetime import datetime 

4from typing import Any 

5 

6from starlette.responses import HTMLResponse, JSONResponse 

7 

8from lexigram.admin.controllers.base import AdminController 

9from lexigram.admin.services.forms.draft_service import ( # type: ignore[import-untyped] 

10 DraftService, 

11) 

12from lexigram.contracts.web import get, post 

13from lexigram.di.decorators import inject 

14from lexigram.logging import get_logger 

15 

16logger = get_logger(__name__) 

17 

18 

19@inject 

20class DraftController(AdminController): 

21 """Controller for handling form auto-save and draft retrieval.""" 

22 

23 def __init__( 

24 self, renderer: Any, draft_service: DraftService | None = None 

25 ) -> None: 

26 super().__init__(renderer) 

27 self.draft_service = draft_service or DraftService() 

28 

29 @post("/api/forms/draft/{form_id}") 

30 async def save_draft(self, form_id: str, request) -> Any: 

31 """Save a form draft via HTMX pulse.""" 

32 try: 

33 # Get user ID from request state 

34 state_user = getattr(getattr(request, "state", None), "user", None) 

35 user_id = ( 

36 str(getattr(state_user, "user_id", "anonymous")) 

37 if state_user 

38 else "anonymous" 

39 ) 

40 

41 # Extract form data 

42 if request.headers.get("content-type") == "application/json": 

43 data = await request.json() 

44 else: 

45 form_data = await request.form() 

46 data = dict(form_data) 

47 

48 # Prevent saving sensitive fields if necessary 

49 data.pop("csrf_token", None) 

50 data.pop("password", None) 

51 

52 await self.draft_service.save_draft(form_id, user_id, data) 

53 

54 # Return success indicator for UI 

55 timestamp = datetime.now().strftime("%H:%M:%S") 

56 return HTMLResponse( 

57 content=f'<span class="text-xs text-muted-foreground italic">Draft saved at {timestamp}</span>', 

58 status_code=200, 

59 ) 

60 

61 except Exception as _save_err: # noqa: BLE001 — controller boundary; autosave errors must not crash the request 

62 logger.exception("Failed to save draft for %s", form_id) 

63 return HTMLResponse( 

64 content='<span class="text-xs text-destructive">Autosave failed</span>', 

65 status_code=500, 

66 ) 

67 

68 @get("/api/forms/draft/{form_id}") 

69 async def get_draft(self, form_id: str, request) -> Any: 

70 """Retrieve a saved form draft.""" 

71 try: 

72 state_user = getattr(getattr(request, "state", None), "user", None) 

73 user_id = ( 

74 str(getattr(state_user, "user_id", "anonymous")) 

75 if state_user 

76 else "anonymous" 

77 ) 

78 draft = await self.draft_service.get_draft(form_id, user_id) 

79 

80 if not draft: 

81 return JSONResponse({"status": "not_found"}, status_code=404) 

82 

83 return JSONResponse( 

84 { 

85 "status": "success", 

86 "data": draft.data, 

87 "updated_at": draft.updated_at.isoformat(), 

88 }, 

89 ) 

90 

91 except Exception as e: # noqa: BLE001 — controller boundary; all errors become HTTP 500 responses 

92 logger.exception("Failed to retrieve draft for %s", form_id) 

93 return JSONResponse({"status": "error", "message": str(e)}, status_code=500)