Coverage for gco / job_envelope.py: 100.00%

1"""The SQS job envelope, shared by the producer and the consumer. 

2 

3``gco jobs submit-sqs`` (:mod:`cli.jobs`) writes this envelope onto the 

4regional queue; the in-cluster queue processor 

5(:mod:`gco.services.queue_processor`) reads it back and applies the manifests. 

6The two halves ship in different artifacts — the CLI runs on an operator's 

7machine, the consumer inside a distroless service image that deliberately 

8cannot import ``cli/`` — so the wire format lived twice: once as a dict 

9literal in the CLI, once as a docstring and a set of ``body.get(...)`` calls 

10in the service. This module is that format, written once. 

11 

12It is deliberately standard-library only, so anything that needs to *speak* 

13the protocol can import it: the service image, the CLI, and CI, which builds a 

14message with this exact code and hands it to the real consumer running in a 

15kind cluster. 

16""" 

17 

18from __future__ import annotations 

19 

20import json 

21import uuid 

22from dataclasses import dataclass 

23from datetime import UTC, datetime 

24from typing import Any 

25 

26#: Namespace recorded in the envelope when neither the caller nor any manifest 

27#: names one. The consumer applies this same fallback when reading a message, 

28#: and it is the namespace the shipped ``job_validation_policy`` allows. 

29DEFAULT_JOB_NAMESPACE = "gco-jobs" 

30 

31 

32def first_manifest_namespace(manifests: list[dict[str, Any]]) -> str | None: 

33 """Return the first explicit ``metadata.namespace`` in a manifest list. 

34 

35 The envelope's ``namespace`` field is informational — the consumer 

36 validates and applies each manifest under the namespace that manifest 

37 declares — so the producer reports the first declared one to keep its 

38 submission response truthful when the caller passes no namespace. 

39 Returns ``None`` when no manifest declares one. 

40 """ 

41 for manifest in manifests: 

42 ns = manifest.get("metadata", {}).get("namespace") if isinstance(manifest, dict) else None 

43 if ns: 

44 return str(ns) 

45 return None 

46 

47 

48@dataclass(frozen=True) 

49class JobMessage: 

50 """One queue submission: the JSON body plus its SQS message attributes.""" 

51 

52 body: dict[str, Any] 

53 attributes: dict[str, dict[str, str]] 

54 

55 @property 

56 def job_id(self) -> str: 

57 """The short id correlating this submission across CLI, queue and logs.""" 

58 return str(self.body["job_id"]) 

59 

60 @property 

61 def namespace(self) -> str: 

62 """The envelope's informational namespace.""" 

63 return str(self.body["namespace"]) 

64 

65 @property 

66 def priority(self) -> int: 

67 """The submitted priority (higher is more important).""" 

68 return int(self.body["priority"]) 

69 

70 def json(self) -> str: 

71 """Serialize the body exactly as it goes onto the queue.""" 

72 return json.dumps(self.body) 

73 

74 

75def build_job_message( 

76 manifests: list[dict[str, Any]], 

77 *, 

78 namespace: str | None = None, 

79 priority: int = 0, 

80 job_id: str | None = None, 

81 submitted_at: datetime | None = None, 

82) -> JobMessage: 

83 """Build the queue message for ``manifests``. 

84 

85 ``job_id`` and ``submitted_at`` default to a fresh short uuid and the 

86 current UTC instant; both are injectable so a caller that needs a 

87 reproducible message (a test, a replay) can pin them. ``namespace`` 

88 falls back to the first namespace the manifests declare, then to 

89 :data:`DEFAULT_JOB_NAMESPACE`. 

90 """ 

91 resolved_id = job_id or str(uuid.uuid4())[:8] 

92 resolved_namespace = namespace or first_manifest_namespace(manifests) or DEFAULT_JOB_NAMESPACE 

93 stamp = submitted_at or datetime.now(UTC) 

94 return JobMessage( 

95 body={ 

96 "job_id": resolved_id, 

97 "manifests": manifests, 

98 "namespace": resolved_namespace, 

99 "priority": priority, 

100 "submitted_at": stamp.isoformat(), 

101 }, 

102 attributes={ 

103 "Priority": {"DataType": "Number", "StringValue": str(priority)}, 

104 "JobId": {"DataType": "String", "StringValue": resolved_id}, 

105 }, 

106 )