Add jetstream consumer priority groups (ADR-42)

nats/src/nats/js/api.py

@@ -67,6 +67,7 @@ class StatusCode(str, Enum):
     NO_MESSAGES = "404"
     REQUEST_TIMEOUT = "408"
     CONFLICT = "409"
+    PIN_ID_MISMATCH = "423"
     CONTROL_MESSAGE = "100"
 
 
@@ -620,6 +621,32 @@ class ReplayPolicy(str, Enum):
     ORIGINAL = "original"
 
 
+class PriorityPolicy(str, Enum):
+    """Priority policy for pull consumer priority groups.
+
+    Enables flexible failover and priority management when multiple clients
+    are pulling from the same consumer.
+
+    Introduced in nats-server 2.11.0 (``PRIORITIZED`` in 2.12.0).
+
+    References:
+        * `Consumers, Pull consumer priority groups <https://docs.nats.io/release-notes/whats_new/whats_new_211#consumers>`
+        * `Consumers, Prioritized pull consumer policy <https://docs.nats.io/release-notes/whats_new/whats_new_212#consumers>`
+    """  # noqa: E501
+
+    NONE = "none"
+    """Default, no priority handling."""
+    PINNED = "pinned_client"
+    """Pins the consumer to a single client per group; others take over when it goes away."""
+    OVERFLOW = "overflow"
+    """Only delivers to a client once ``min_pending`` or ``min_ack_pending`` thresholds are reached."""
+    PRIORITIZED = "prioritized"
+    """Delivers to the client with the highest priority (0-9, 0 is highest and default).
+
+    Introduced in nats-server 2.12.0.
+    """
+
+
 @dataclass
 class ConsumerConfig(Base):
     """Consumer configuration.
@@ -672,12 +699,26 @@ class ConsumerConfig(Base):
     # Introduced in nats-server 2.11.0.
     pause_until: Optional[str] = None
 
+    # Priority policy.
+    # Introduced in nats-server 2.11.0.
+    priority_policy: Optional[PriorityPolicy] = None
+
+    # The duration (seconds) after which the client will be unpinned if no new
+    # pull requests are sent. Used with PriorityPolicy.PINNED.
+    # Introduced in nats-server 2.11.0.
+    priority_timeout: Optional[float] = None
+
+    # Priority groups this consumer supports.
+    # Introduced in nats-server 2.11.0.
+    priority_groups: Optional[list[str]] = None
+
     @classmethod
     def from_response(cls, resp: Dict[str, Any]):
         cls._convert_nanoseconds(resp, "ack_wait")
         cls._convert_nanoseconds(resp, "idle_heartbeat")
         cls._convert_nanoseconds(resp, "inactive_threshold")
         cls._convert_utc_iso(resp, "opt_start_time")
+        cls._convert_nanoseconds(resp, "priority_timeout")
         if "backoff" in resp:
             resp["backoff"] = [val / _NANOSECOND for val in resp["backoff"]]
         return super().from_response(resp)
@@ -689,6 +730,8 @@ def as_dict(self) -> Dict[str, object]:
         result["ack_wait"] = self._to_nanoseconds(self.ack_wait)
         result["idle_heartbeat"] = self._to_nanoseconds(self.idle_heartbeat)
         result["inactive_threshold"] = self._to_nanoseconds(self.inactive_threshold)
+        if self.priority_timeout is not None:
+            result["priority_timeout"] = self._to_nanoseconds(self.priority_timeout)
         if self.backoff:
             result["backoff"] = [self._to_nanoseconds(i) for i in self.backoff]
         return result
@@ -712,6 +755,32 @@ def as_dict(self) -> Dict[str, object]:
         return result
 
 
+@dataclass
+class PriorityGroupState(Base):
+    """
+    State of a consumer priority group.
+
+    Introduced in nats-server 2.11.0.
+    """
+
+    group: str
+    # Generated ID of the pinned client. Only set when a client is pinned.
+    pinned_client_id: Optional[str] = None
+    # When the client was pinned. Only set when a client is pinned.
+    pinned_ts: Optional[datetime.datetime] = None
+
+    @classmethod
+    def from_response(cls, resp: Dict[str, Any]):
+        cls._convert_utc_iso(resp, "pinned_ts")
+        return super().from_response(resp)
+
+    def as_dict(self) -> Dict[str, object]:
+        result = super().as_dict()
+        if self.pinned_ts is not None:
+            result["pinned_ts"] = self._to_utc_iso(self.pinned_ts)
+        return result
+
+
 @dataclass
 class ConsumerInfo(Base):
     """
@@ -736,13 +805,16 @@ class ConsumerInfo(Base):
     # RFC 3339 timestamp until which the consumer is paused.
     # Introduced in nats-server 2.11.0.
     pause_remaining: Optional[str] = None
+    # Introduced in nats-server 2.11.0.
+    priority_groups: Optional[list[PriorityGroupState]] = None
 
     @classmethod
     def from_response(cls, resp: Dict[str, Any]):
         cls._convert(resp, "delivered", SequenceInfo)
         cls._convert(resp, "ack_floor", SequenceInfo)
         cls._convert(resp, "config", ConsumerConfig)
         cls._convert(resp, "cluster", ClusterInfo)
+        cls._convert(resp, "priority_groups", PriorityGroupState)
         cls._convert_utc_iso(resp, "created")
         return super().from_response(resp)