diff --git a/nats-jetstream/src/nats/jetstream/__init__.py b/nats-jetstream/src/nats/jetstream/__init__.py index a156ae0b..a8e07642 100644 --- a/nats-jetstream/src/nats/jetstream/__init__.py +++ b/nats-jetstream/src/nats/jetstream/__init__.py @@ -8,6 +8,7 @@ from datetime import datetime from typing import TYPE_CHECKING, AsyncIterator, overload +from nats.client.errors import NoRespondersError from nats.client.message import Headers from nats.client.protocol.message import parse_headers from nats.jetstream import api @@ -313,8 +314,6 @@ async def publish( """ import asyncio - from nats.client.errors import NoRespondersError - # Track overall deadline start_time = asyncio.get_event_loop().time() deadline = start_time + timeout @@ -458,7 +457,14 @@ async def create_stream(self, config: StreamConfig | None = None, /, **kwargs) - # Convert StreamConfig to API request format and create stream config_dict = config.to_request() - response = await self._api.stream_create(config.name, **config_dict) + try: + response = await self._api.stream_create(config.name, **config_dict) + except JetStreamError as e: + if e.error_code == ErrorCode.STREAM_NAME_IN_USE: + raise StreamNameAlreadyInUseError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise info = StreamInfo.from_response(response, strict=self._strict) return Stream(self, config.name, info) @@ -479,7 +485,14 @@ async def update_stream(self, **config) -> StreamInfo: name = config.get("name") if name is None: raise ValueError("Stream name is required for update") - response = await self._api.stream_update(name, **config) + try: + response = await self._api.stream_update(name, **config) + except JetStreamError as e: + if e.error_code == ErrorCode.STREAM_NOT_FOUND: + raise StreamNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return StreamInfo.from_response(response, strict=self._strict) async def delete_stream(self, name: str) -> bool: @@ -495,7 +508,14 @@ async def delete_stream(self, name: str) -> bool: StreamNotFoundError: If the stream does not exist JetStreamError: For other JetStream API errors """ - response = await self._api.stream_delete(name) + try: + response = await self._api.stream_delete(name) + except JetStreamError as e: + if e.error_code == ErrorCode.STREAM_NOT_FOUND: + raise StreamNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return response["success"] async def get_stream_info( @@ -507,12 +527,19 @@ async def get_stream_info( offset: int | None = None, ) -> StreamInfo: """Get information about a stream.""" - response = await self._api.stream_info( - name, - deleted_details=deleted_details, - subjects_filter=subjects_filter, - offset=offset, - ) + try: + response = await self._api.stream_info( + name, + deleted_details=deleted_details, + subjects_filter=subjects_filter, + offset=offset, + ) + except JetStreamError as e: + if e.error_code == ErrorCode.STREAM_NOT_FOUND: + raise StreamNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return StreamInfo.from_response(response, strict=self._strict) async def get_stream(self, name: str) -> Stream: @@ -743,7 +770,14 @@ async def get_consumer_info(self, stream_name: str, consumer_name: str) -> Consu Returns: Consumer information """ - response = await self._api.consumer_info(stream_name, consumer_name) + try: + response = await self._api.consumer_info(stream_name, consumer_name) + except JetStreamError as e: + if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: + raise ConsumerNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return ConsumerInfo.from_response(response, strict=self._strict) async def account_info(self) -> AccountInfo: @@ -757,7 +791,23 @@ async def account_info(self) -> AccountInfo: JetStreamNotEnabledForAccountError: If JetStream is not enabled for this account JetStreamError: For other JetStream API errors """ - response = await self._api.account_info() + try: + response = await self._api.account_info() + except NoRespondersError as e: + # No responders means JetStream is not enabled on the server. + raise JetStreamNotEnabledError( + "JetStream not enabled", code=503, error_code=ErrorCode.JETSTREAM_NOT_ENABLED + ) from e + except JetStreamError as e: + if e.error_code == ErrorCode.JETSTREAM_NOT_ENABLED_FOR_ACCOUNT: + raise JetStreamNotEnabledForAccountError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + if e.error_code == ErrorCode.JETSTREAM_NOT_ENABLED: + raise JetStreamNotEnabledError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return AccountInfo.from_response(response, strict=self._strict) async def get_message(self, stream: str, sequence: int) -> StreamMessage: @@ -777,7 +827,14 @@ async def get_message(self, stream: str, sequence: int) -> StreamMessage: MessageNotFoundError: If the message does not exist JetStreamError: For other JetStream API errors """ - response = await self._api.stream_msg_get(stream, seq=sequence) + try: + response = await self._api.stream_msg_get(stream, seq=sequence) + except JetStreamError as e: + if e.error_code == ErrorCode.MESSAGE_NOT_FOUND: + raise MessageNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise message = response["message"] # Decode base64 data if present @@ -819,7 +876,14 @@ async def get_last_message_for_subject(self, stream: str, subject: str) -> Strea Returns: The stream message including subject, data, headers, etc. """ - response = await self._api.stream_msg_get(stream, last_by_subj=subject) + try: + response = await self._api.stream_msg_get(stream, last_by_subj=subject) + except JetStreamError as e: + if e.error_code == ErrorCode.MESSAGE_NOT_FOUND: + raise MessageNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise message = response["message"] # Decode base64 data if present diff --git a/nats-jetstream/src/nats/jetstream/api/client.py b/nats-jetstream/src/nats/jetstream/api/client.py index 6cf4d9e9..6af0cdfb 100644 --- a/nats-jetstream/src/nats/jetstream/api/client.py +++ b/nats-jetstream/src/nats/jetstream/api/client.py @@ -13,19 +13,8 @@ overload, ) -from nats.client.errors import NoRespondersError - from ..errors import ( - ConsumerInvalidResetError, - ConsumerNotFoundError, - ErrorCode, JetStreamError, - JetStreamNotEnabledError, - JetStreamNotEnabledForAccountError, - MaximumConsumersLimitError, - MessageNotFoundError, - StreamNameAlreadyInUseError, - StreamNotFoundError, ) from .types import ( AccountInfoResponse, @@ -131,72 +120,31 @@ def __init__( self._raise_on_unknown_keys = raise_on_unknown_keys async def account_info(self) -> AccountInfoResponse: - try: - return await self.request_json( - f"{self._prefix}.INFO", - response_type=AccountInfoResponse, - ) - except NoRespondersError as e: - # If no responders, JetStream is not enabled on the server - raise JetStreamNotEnabledError( - "JetStream not enabled", code=503, error_code=ErrorCode.JETSTREAM_NOT_ENABLED - ) from e - except JetStreamError as e: - if e.error_code == ErrorCode.JETSTREAM_NOT_ENABLED_FOR_ACCOUNT: - raise JetStreamNotEnabledForAccountError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - if e.error_code == ErrorCode.JETSTREAM_NOT_ENABLED: - raise JetStreamNotEnabledError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.INFO", + response_type=AccountInfoResponse, + ) async def consumer_create( self, stream_name: str, consumer_name: str, /, **request: Unpack[ConsumerCreateRequest] ) -> ConsumerCreateResponse: - try: - return await self.request_json( - f"{self._prefix}.CONSUMER.CREATE.{stream_name}.{consumer_name}", - request, - response_type=ConsumerCreateResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.STREAM_NOT_FOUND: - raise StreamNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - if e.error_code == ErrorCode.MAXIMUM_CONSUMERS_LIMIT: - raise MaximumConsumersLimitError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.CONSUMER.CREATE.{stream_name}.{consumer_name}", + request, + response_type=ConsumerCreateResponse, + ) async def consumer_delete(self, stream_name: str, consumer_name: str, /) -> ConsumerDeleteResponse: - try: - return await self.request_json( - f"{self._prefix}.CONSUMER.DELETE.{stream_name}.{consumer_name}", - response_type=ConsumerDeleteResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: - raise ConsumerNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.CONSUMER.DELETE.{stream_name}.{consumer_name}", + response_type=ConsumerDeleteResponse, + ) async def consumer_info(self, stream_name: str, consumer_name: str, /) -> ConsumerInfoResponse: - try: - return await self.request_json( - f"{self._prefix}.CONSUMER.INFO.{stream_name}.{consumer_name}", - response_type=ConsumerInfoResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: - raise ConsumerNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.CONSUMER.INFO.{stream_name}.{consumer_name}", + response_type=ConsumerInfoResponse, + ) async def consumer_list(self, stream_name: str, /, **request: Unpack[ConsumerListRequest]) -> ConsumerListResponse: """Get information about all consumers in a stream.""" @@ -229,18 +177,11 @@ async def consumer_pause( Returns: ConsumerPauseResponse with pause state """ - try: - return await self.request_json( - f"{self._prefix}.CONSUMER.PAUSE.{stream_name}.{consumer_name}", - request if request else None, - response_type=ConsumerPauseResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: - raise ConsumerNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.CONSUMER.PAUSE.{stream_name}.{consumer_name}", + request if request else None, + response_type=ConsumerPauseResponse, + ) async def consumer_reset( self, stream_name: str, consumer_name: str, /, **request: Unpack[ConsumerResetRequest] @@ -260,65 +201,31 @@ async def consumer_reset( ConsumerResetResponse with refreshed consumer state and the stream sequence the consumer was reset to. """ - try: - return await self.request_json( - f"{self._prefix}.CONSUMER.RESET.{stream_name}.{consumer_name}", - request if request else None, - response_type=ConsumerResetResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: - raise ConsumerNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - if e.error_code == ErrorCode.CONSUMER_INVALID_RESET: - raise ConsumerInvalidResetError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.CONSUMER.RESET.{stream_name}.{consumer_name}", + request if request else None, + response_type=ConsumerResetResponse, + ) async def stream_create(self, name: str, /, **request: Unpack[StreamCreateRequest]) -> StreamCreateResponse: - try: - return await self.request_json( - f"{self._prefix}.STREAM.CREATE.{name}", - request, - response_type=StreamCreateResponse, - ) - except JetStreamError as e: - # Re-raise specific errors (matching Go's error handling) - if e.error_code == ErrorCode.STREAM_NAME_IN_USE: - raise StreamNameAlreadyInUseError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - # Unknown errors pass through as generic JetStreamError - raise + return await self.request_json( + f"{self._prefix}.STREAM.CREATE.{name}", + request, + response_type=StreamCreateResponse, + ) async def stream_delete(self, name: str, /) -> StreamDeleteResponse: - try: - return await self.request_json( - f"{self._prefix}.STREAM.DELETE.{name}", - response_type=StreamDeleteResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.STREAM_NOT_FOUND: - raise StreamNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.STREAM.DELETE.{name}", + response_type=StreamDeleteResponse, + ) async def stream_info(self, name: str, /, **request: Unpack[StreamInfoRequest]) -> StreamInfoResponse: - try: - return await self.request_json( - f"{self._prefix}.STREAM.INFO.{name}", - request if request else None, - response_type=StreamInfoResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.STREAM_NOT_FOUND: - raise StreamNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.STREAM.INFO.{name}", + request if request else None, + response_type=StreamInfoResponse, + ) async def stream_list(self, **request: Unpack[StreamListRequest]) -> StreamListResponse: """Get information about all streams. @@ -345,18 +252,11 @@ async def stream_msg_delete( ) async def stream_msg_get(self, name: str, /, **request: Unpack[StreamMsgGetRequest]) -> StreamMsgGetResponse: - try: - return await self.request_json( - f"{self._prefix}.STREAM.MSG.GET.{name}", - request if request else None, - response_type=StreamMsgGetResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.MESSAGE_NOT_FOUND: - raise MessageNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.STREAM.MSG.GET.{name}", + request if request else None, + response_type=StreamMsgGetResponse, + ) async def stream_names(self, **request: Unpack[StreamNamesRequest]) -> StreamNamesResponse: """Get a list of all stream names. @@ -381,18 +281,11 @@ async def stream_purge(self, name: str, /, **request: Unpack[StreamPurgeRequest] ) async def stream_update(self, name: str, /, **request: Unpack[StreamUpdateRequest]) -> StreamUpdateResponse: - try: - return await self.request_json( - f"{self._prefix}.STREAM.UPDATE.{name}", - request, - response_type=StreamUpdateResponse, - ) - except JetStreamError as e: - if e.error_code == ErrorCode.STREAM_NOT_FOUND: - raise StreamNotFoundError( - e.description, code=e.code, error_code=e.error_code, description=e.description - ) from e - raise + return await self.request_json( + f"{self._prefix}.STREAM.UPDATE.{name}", + request, + response_type=StreamUpdateResponse, + ) @overload async def request_json[ResponseT]( diff --git a/nats-jetstream/src/nats/jetstream/stream.py b/nats-jetstream/src/nats/jetstream/stream.py index 462bc742..78da29bd 100644 --- a/nats-jetstream/src/nats/jetstream/stream.py +++ b/nats-jetstream/src/nats/jetstream/stream.py @@ -22,7 +22,15 @@ from .consumer import Consumer, ConsumerConfig, ConsumerInfo, ConsumerReset, OrderedConsumerConfig from .consumer.ordered import OrderedConsumer from .consumer.pull import PullConsumer -from .errors import MessageNotFoundError +from .errors import ( + ConsumerInvalidResetError, + ConsumerNotFoundError, + ErrorCode, + JetStreamError, + MaximumConsumersLimitError, + MessageNotFoundError, + StreamNotFoundError, +) CONSUMER_ACTION_CREATE = "create" CONSUMER_ACTION_UPDATE = "update" @@ -1242,7 +1250,14 @@ async def _get_message(self, *, sequence: int | None = None, last_by_subject: st # }, # "type": "io.nats.jetstream.api.v1.stream_msg_get_response" # } - response = await api.stream_msg_get(self._name, seq=sequence, last_by_subj=last_by_subject) + try: + response = await api.stream_msg_get(self._name, seq=sequence, last_by_subj=last_by_subject) + except JetStreamError as e: + if e.error_code == ErrorCode.MESSAGE_NOT_FOUND: + raise MessageNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise message = response["message"] # Decode base64 data if present @@ -1369,7 +1384,18 @@ async def _upsert_consumer(self, action: str = CONSUMER_ACTION_CREATE_OR_UPDATE, } # Create/update consumer via API - response = await api.consumer_create(self._name, name, **request) + try: + response = await api.consumer_create(self._name, name, **request) + except JetStreamError as e: + if e.error_code == ErrorCode.STREAM_NOT_FOUND: + raise StreamNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + if e.error_code == ErrorCode.MAXIMUM_CONSUMERS_LIMIT: + raise MaximumConsumersLimitError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise consumer_info = ConsumerInfo.from_response(response, strict=self._jetstream.strict) # Check if this is a push consumer (has deliver_subject) @@ -1437,7 +1463,14 @@ async def get_consumer_info(self, consumer_name: str) -> ConsumerInfo: raise RuntimeError("JetStream does not have an API client") # Get consumer info via API - response = await api.consumer_info(self._name, consumer_name) + try: + response = await api.consumer_info(self._name, consumer_name) + except JetStreamError as e: + if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: + raise ConsumerNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return ConsumerInfo.from_response(response, strict=self._jetstream.strict) async def get_consumer(self, consumer_name: str) -> Consumer: @@ -1483,7 +1516,14 @@ async def delete_consumer(self, consumer_name: str) -> bool: raise RuntimeError("JetStream does not have an API client") # Delete consumer via API - response = await api.consumer_delete(self._name, consumer_name) + try: + response = await api.consumer_delete(self._name, consumer_name) + except JetStreamError as e: + if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: + raise ConsumerNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return response["success"] @overload @@ -1581,7 +1621,14 @@ async def pause_consumer(self, consumer_name: str, pause_until: float) -> None: pause_until_str = dt.isoformat().replace("+00:00", "Z") # Pause consumer via API - await api.consumer_pause(self._name, consumer_name, pause_until=pause_until_str) + try: + await api.consumer_pause(self._name, consumer_name, pause_until=pause_until_str) + except JetStreamError as e: + if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: + raise ConsumerNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise async def resume_consumer(self, consumer_name: str) -> None: """Resume a paused consumer immediately. @@ -1596,7 +1643,14 @@ async def resume_consumer(self, consumer_name: str) -> None: # Resume by setting pause_until to a time in the past (epoch) # RFC3339 format: "1970-01-01T00:00:00Z" - await api.consumer_pause(self._name, consumer_name) + try: + await api.consumer_pause(self._name, consumer_name) + except JetStreamError as e: + if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: + raise ConsumerNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise async def reset_consumer(self, consumer_name: str, seq: int | None = None) -> ConsumerReset: """Reset a consumer's delivery state (ADR-60). @@ -1634,10 +1688,21 @@ async def reset_consumer(self, consumer_name: str, seq: int | None = None) -> Co if api is None: raise RuntimeError("JetStream does not have an API client") - if seq is None: - response = await api.consumer_reset(self._name, consumer_name) - else: - response = await api.consumer_reset(self._name, consumer_name, seq=seq) + try: + if seq is None: + response = await api.consumer_reset(self._name, consumer_name) + else: + response = await api.consumer_reset(self._name, consumer_name, seq=seq) + except JetStreamError as e: + if e.error_code == ErrorCode.CONSUMER_NOT_FOUND: + raise ConsumerNotFoundError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + if e.error_code == ErrorCode.CONSUMER_INVALID_RESET: + raise ConsumerInvalidResetError( + e.description, code=e.code, error_code=e.error_code, description=e.description + ) from e + raise return ConsumerReset.from_response(response, strict=self._jetstream.strict) @overload