Skip to content

Map JetStream errors in callers, not the API client #984

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
caspervonb wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Map JetStream errors in callers, not the API client #984

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8093d1f
Map JetStream errors in callers, not the API client
caspervonb Jun 22, 2026
e36705c
Import NoRespondersError at module level
caspervonb Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 79 additions & 15 deletions nats-jetstream/src/nats/jetstream/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand All @@ -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:
Expand All @@ -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(
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand All @@ -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
"""

@synadia-claude-reviewer synadia-claude-reviewer Bot Jun 22, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The deferred import of NoRespondersError inside the method body is non-idiomatic. Since nats.client is already a module-level dependency (e.g. from nats.client.message import Headers at the top of the file), there is no circular import risk here. This should sit with the other nats.client imports at the top of the file.

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:
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading