Coverage for src/lumen/api/_helpers.py: 86%

1""" 

2This file contains shared helper utilities for HTTP requests and responses. 

3It provides functions for parsing headers, handling locations, and structuring outcomes. 

4""" 

5 

6from __future__ import annotations 

7 

8from datetime import UTC, datetime 

9from email.utils import format_datetime 

10from typing import Any 

11 

12import uuid_utils 

13 

14from ..core.exceptions import OperationOutcomeError 

15 

16 

17def etag(version_id: int) -> str: 

18 return f'W/"{version_id}"' 

19 

20 

21def last_modified(dt: datetime) -> str: 

22 """HTTP-date for the ``Last-Modified`` response header (RFC 7231).""" 

23 if dt.tzinfo is None: 

24 dt = dt.replace(tzinfo=UTC) 

25 return format_datetime(dt, usegmt=True) 

26 

27 

28def last_modified_instant(dt: datetime) -> str: 

29 """FHIR ``instant`` (ISO 8601) for ``Bundle.entry.response.lastModified``. 

30 

31 This is a FHIR data field, not an HTTP header — it must be ISO 8601, not 

32 the RFC 7231 HTTP-date produced by :func:`last_modified`. 

33 """ 

34 if dt.tzinfo is None: 

35 dt = dt.replace(tzinfo=UTC) 

36 return dt.isoformat() 

37 

38 

39def inject_meta( 

40 resource: dict[str, Any], version_id: int, last_updated: datetime 

41) -> dict[str, Any]: 

42 """Return resource with meta.versionId and meta.lastUpdated set from DB values.""" 

43 meta: dict[str, Any] = dict(resource.get("meta") or {}) 

44 meta["versionId"] = str(version_id) 

45 if last_updated.tzinfo is None: 

46 last_updated = last_updated.replace(tzinfo=UTC) 

47 meta["lastUpdated"] = last_updated.isoformat() 

48 return {**resource, "meta": meta} 

49 

50 

51def new_id() -> str: 

52 return str(uuid_utils.uuid7()) 

53 

54 

55def parse_if_match(header: str | None) -> int | None: 

56 """Parse If-Match: W/"<version_id>" → int, or None if absent.""" 

57 if header is None: 

58 return None 

59 stripped = header.strip() 

60 if stripped.startswith('W/"') and stripped.endswith('"'): 

61 try: 

62 return int(stripped[3:-1]) 

63 except ValueError: 

64 pass 

65 raise OperationOutcomeError(400, "invalid", f"Malformed If-Match header: {header}")