Coverage for domain/result.py: 76%

1"""Tagged error algebra + Result, ported from ``arive-spec/src/domain/result.ts``. 

2 

3Pure: no Django, no HTTP. The API edge converts a ``DomainError`` into the shared 

4error envelope. 

5""" 

6 

7from __future__ import annotations 

8 

9import uuid 

10from dataclasses import dataclass, field 

11from datetime import UTC, datetime 

12from typing import Any 

13 

14 

15@dataclass 

16class AppError: 

17 """A business-rule (``ORG-xxxx``) or generic error.""" 

18 

19 code: str 

20 message: str 

21 field: str | None = None 

22 details: str | None = None 

23 

24 

25@dataclass 

26class Result: 

27 ok: bool 

28 value: Any = None 

29 errors: list[AppError] = field(default_factory=list) 

30 

31 

32def ok(value: Any = None) -> Result: 

33 return Result(True, value, []) 

34 

35 

36def err(*errors: AppError) -> Result: 

37 return Result(False, None, list(errors)) 

38 

39 

40def all_(results: list[Result]) -> Result: 

41 """Combine results; collects all errors (does not short-circuit).""" 

42 collected = [e for r in results if not r.ok for e in r.errors] 

43 if collected: 

44 return err(*collected) 

45 return ok([r.value for r in results]) 

46 

47 

48def http_status_for(errors: list[AppError]) -> int: 

49 """Map error codes to HTTP status. Note: unlike the legacy ARIVE API (which 

50 returned 500 for ORG-xxxx), business-rule violations map to 422 here.""" 

51 codes = {e.code for e in errors} 

52 if "NOT_FOUND" in codes: 

53 return 404 

54 if "UNAUTHORIZED" in codes: 

55 return 401 

56 if "FORBIDDEN" in codes: 

57 return 403 

58 if "INVALID_TRANSITION" in codes or "CONFLICT" in codes: 

59 return 409 

60 return 422 

61 

62 

63def to_envelope(status_code: int, errors: list[AppError], ref: str | None = None) -> dict: 

64 """Standard API error envelope: {statusCode, message, details, ref, ...}.""" 

65 return { 

66 "statusCode": status_code, 

67 "message": errors[0].message if errors else "Request failed", 

68 "details": "; ".join(f"{e.code}: {e.message}" for e in errors), 

69 "timestamp": datetime.now(UTC).isoformat(), 

70 "ref": ref or ("req_" + uuid.uuid4().hex[:8]), 

71 "errors": [vars(e) for e in errors], 

72 } 

73 

74 

75def format_db_error(exc: Exception, model_name: str = "") -> str: 

76 """Format a database exception into a user-friendly error message without leaking SQL or Python tracebacks.""" 

77 msg = str(exc).lower() 

78 if "numeric" in msg and "overflow" in msg: 

79 return "A numeric value is out of range. Please enter a valid number within allowed bounds." 

80 if "value too long" in msg: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true

81 return "One or more text fields exceed the maximum allowed length." 

82 if "unique constraint" in msg or "duplicate key" in msg: 82 ↛ 84line 82 didn't jump to line 84 because the condition on line 82 was always true

83 return f"A record with these details already exists in {model_name or 'the system'}." 

84 if "foreign key constraint" in msg: 

85 return "The referenced related record does not exist or has been removed." 

86 if "null value" in msg and "violates not-null" in msg: 

87 return "A required field was left empty." 

88 return f"Unable to save {model_name or 'record'} due to invalid data. Please check your inputs and try again." 

89 

90 

91class DomainError(Exception): 

92 """Raised at the API boundary; rendered as the error envelope.""" 

93 

94 def __init__(self, errors: list[AppError], status_code: int | None = None): 

95 self.errors = errors 

96 self.status_code = status_code or http_status_for(errors) 

97 super().__init__(errors[0].message if errors else "Request failed")