diff --git a/CLAUDE.md b/CLAUDE.md index 629bbfd..8da2f81 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,9 +44,21 @@ The document category deliberately has **no `base.py` ABC and no backend wrapper AWS `from_assume_role` (STS AssumeRole, cross-account) is wired for **SQS** (`get_queue`) and **S3** (`get_storage` + `get_storage_client`): `role_arn` present → assume-role path, checked **before** `aws_access_key_id`. It does a synchronous `boto3` `sts:AssumeRole` (optional `external_id` → `ExternalId`) and threads the temporary creds into the `aioboto3.Session`. Temp creds are not auto-refreshed — construct a fresh backend when the session expires. -### Messaging payload contract (byte-primitive) +### Azure credentials — one chain, defined once -The messaging primitive is **raw bytes + a flat `attributes` map** (string → string), not JSON. `MessagingBackend.send(body: bytes, attributes=None, delay=0, *, group_id, dedup_id)` and `send_batch(messages: list[OutgoingMessage], ...)` where `OutgoingMessage` (in `base.py`) is `{body: bytes, attributes: dict[str,str] | None}`. Attributes map to SQS `MessageAttributes` (String type) and Service Bus `application_properties`. Received `Message.body` is `bytes`; `Message.attributes` is the stringified provider attributes. JSON convenience: the module-level `send_json(backend, message: dict, attributes=None, delay=0, ...)` helper (works for any backend — `json.dumps(...).encode()` then `send`) plus `Message.json()` (decodes the body with `json.loads`). Receipt-handle / ack / FIFO semantics are unchanged. +Every Azure `from_managed_identity` builds its credential through **`cloudrift/core/azure_credentials.py`** (`build_async_credential` / `build_credential`), never by instantiating an `azure.identity` class directly. The chain is `DefaultAzureCredential` with the developer-machine sources excluded, yielding **workload identity → managed identity → az CLI**. Excluding `environment` is the point, not an accident: ambient `AZURE_CLIENT_ID`/`AZURE_CLIENT_SECRET` would otherwise shadow the workload's real identity — the same reasoning as SQS's `exclude_env_credentials`. When adding an Azure backend, call the helper; **do not** reintroduce `ManagedIdentityCredential`. + +Each constructor takes `credential_options: dict | None = None`, forwarded verbatim to `DefaultAzureCredential` (overrides win over the defaults). It is deliberately an explicit dict rather than `**kwargs`: with `**kwargs` a misspelled *backend* option (e.g. `session_enable` for `session_enabled`) would be silently swallowed as a credential option instead of raising `TypeError`. Two backends pass it positionally-adjacent to their own `**kwargs` (`crypto`, `sql/mssql`), which is the same reason. `azure-identity` is pinned `>=1.15.0`, which predates `exclude_broker_credential` — so that one is not set by default; callers on newer releases can pass it through `credential_options`. + +Credentials are **not** shared between backends: each owns the one it built and closes it in `close()`, so a module-level singleton would let one backend's shutdown break another's. + +### Messaging payload contract (dict-primitive) + +The messaging primitive is a **dict + a flat `attributes` map** (string → string). Unlike every other category, `send`/`send_batch` are **concrete on the ABC**, not abstract: `MessagingBackend.send(payload: dict, attributes=None, delay=0, *, group_id, dedup_id)` serializes through the module-level `to_json(payload, default=...)` — the single serialization point — and calls the backend's `_send_json(body: str, ...)`. `send_batch(messages: list[OutgoingMessage], ...)` where `OutgoingMessage` is `{body: dict, attributes: dict[str,str] | None}` likewise calls `_send_json_batch(items: list[tuple[str, dict|None]], ...)`. **When adding a backend, implement `_send_json`/`_send_json_batch` and never override `send`/`send_batch`** — that is what keeps serialization uniform. `to_json` rejects non-dicts with `TypeError`, so a backend can never see an unserialized payload and a caller can never send raw bytes. The class attribute `json_default` (default `str`) is the `json.dumps` fallback for `datetime`/`Decimal`/`UUID`; override it per backend if fidelity matters. + +The hook passes a **`str`, not `bytes`**, because neither provider wants bytes: SQS's `MessageBody` is typed `string` (so bytes would force an `encode`/`decode` round-trip), and `ServiceBusMessage(str)` produces a byte-identical AMQP `DATA` body to `ServiceBusMessage(bytes)`. + +Attributes map to SQS `MessageAttributes` (String type) and Service Bus `application_properties`. On the **receive** side `Message.body` is still `bytes` — deliberately asymmetric, so a malformed or non-UTF-8 payload from a foreign producer stays inspectable for DLQ triage — with `Message.data` as the `json.loads`'d dict. `Message.attributes` is the stringified provider attributes. Receipt-handle / ack / FIFO semantics are unchanged. ### Redis cache specifics diff --git a/README.md b/README.md index 43fdb87..22a6cc4 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,43 @@ storage = get_storage( ) ``` +### Azure authentication + +Every Azure backend resolves its identity through the same chain, so one code +path works in all three environments a service runs in: + +``` +Workload Identity → Managed Identity → Azure CLI + (AKS) (App Service / (local dev, + Container Apps / after `az login`) + VM) +``` + +That means `get_storage("azure_blob", account_url=...)` with no credentials works +unchanged on AKS, on App Service, and on your laptop. Pass `client_id=` to select a +*user-assigned* managed identity; omit it for the system-assigned one. + +Developer-machine credential sources are deliberately excluded: ambient +`AZURE_CLIENT_ID` / `AZURE_CLIENT_SECRET` env vars (which would silently shadow the +workload's real identity — the Azure counterpart of SQS's `exclude_env_credentials`), +plus the shared token cache, VS Code, PowerShell, and `azd`. + +To override — for example to lock a production service down to managed identity only: + +```python +storage = get_storage("azure_blob", account_url="...", container="c", + credential_options={"exclude_cli_credential": True}) +``` + +`credential_options` is accepted by every Azure `from_managed_identity` constructor and +is forwarded verbatim to `DefaultAzureCredential`. The chain itself is defined once in +`cloudrift/core/azure_credentials.py`. + +> On a developer machine this means code intended to exercise managed identity will +> instead succeed using your `az login` identity rather than failing. Pass +> `credential_options={"exclude_cli_credential": True}` when you need to test the +> production path. + --- ## Storage @@ -146,26 +183,22 @@ bus = get_queue("azure_bus", fully_qualified_namespace="ns.servicebus.windows.ne queue_name="my-queue") # managed identity ``` -**Operations** — the payload primitive is **raw bytes** plus an optional flat -`attributes` map (string → string). Attributes map to SQS `MessageAttributes` +**Operations** — pass a **dict**; cloudrift serializes it to JSON. Optionally add a +flat `attributes` map (string → string), which maps to SQS `MessageAttributes` (String type) / Service Bus `application_properties`: ```python -from cloudrift.messaging import OutgoingMessage, send_json +from cloudrift.messaging import OutgoingMessage -# Raw bytes + attributes -msg_id = await queue.send(b"raw payload", attributes={"content_type": "text/plain"}) +msg_id = await queue.send({"action": "process", "id": 42}, attributes={"v": "1"}) ids = await queue.send_batch([ - OutgoingMessage(body=b"a", attributes={"k": "1"}), - OutgoingMessage(body=b"b"), + OutgoingMessage(body={"n": 1}, attributes={"k": "1"}), + OutgoingMessage(body={"n": 2}), ]) -# JSON convenience: send_json() dumps+encodes, m.json() loads the bytes body -await send_json(queue, {"action": "process", "id": 42}, attributes={"v": "1"}) - messages = await queue.receive(max_messages=10, wait_time=20) # long-poll for m in messages: - handle_job(m.json()) # or m.body for the raw bytes + handle_job(m.data) # the payload dict; m.body is the raw bytes print(m.attributes) # str -> str map await queue.delete(m.receipt_handle) # ack # or: await queue.nack(m.receipt_handle) # return for immediate redelivery @@ -174,6 +207,14 @@ await queue.purge() await queue.close() ``` +`send()` takes a `dict` and nothing else — passing bytes, a string, or a list raises +`TypeError`. Values `json.dumps` can't encode natively (`datetime`, `Decimal`, `UUID`) +are stringified via the backend's `json_default`, which defaults to `str`. + +`Message.body` stays **raw bytes** on the receive path so a malformed or non-UTF-8 +payload from a foreign producer is still inspectable for dead-letter triage; +`Message.data` is the decoded dict. + > **Azure Service Bus note:** receipt handles are lock tokens — they are only > valid on the same backend instance that received the message, and only within > the message lock duration. SQS receipt handles, by contrast, are plain strings @@ -188,13 +229,13 @@ queues, pass `group_id` (ordering key) and `dedup_id` (deduplication key): # SQS FIFO — group_id is required, dedup_id optional if the queue has # content-based deduplication enabled fifo = get_queue("sqs", queue_url="https://sqs.../jobs.fifo", region="us-east-1") -await send_json(fifo, {"task": "extract"}, group_id="owner-123", dedup_id="evt-abc") +await fifo.send({"task": "extract"}, group_id="owner-123", dedup_id="evt-abc") # Azure Service Bus — queue must be created with sessions enabled; # pass session_enabled=True so the backend uses session receivers bus = get_queue("azure_bus", connection_string="...", queue_name="jobs", session_enabled=True) -await send_json(bus, {"task": "extract"}, group_id="owner-123", dedup_id="evt-abc") +await bus.send({"task": "extract"}, group_id="owner-123", dedup_id="evt-abc") messages = await fifo.receive(max_messages=10, wait_time=20, visibility_timeout=300) for m in messages: diff --git a/cloudrift/__init__.py b/cloudrift/__init__.py index d20879c..ef82907 100644 --- a/cloudrift/__init__.py +++ b/cloudrift/__init__.py @@ -8,7 +8,7 @@ from cloudrift.pubsub import get_pubsub from cloudrift.email import get_email -__version__ = "0.2.6" +__version__ = "0.2.7" __all__ = [ "get_storage", "get_queue", diff --git a/cloudrift/cache/redis_azure.py b/cloudrift/cache/redis_azure.py index 6d5b44c..899fcf6 100644 --- a/cloudrift/cache/redis_azure.py +++ b/cloudrift/cache/redis_azure.py @@ -64,24 +64,29 @@ def from_managed_identity( ssl: bool = True, client_id: str | None = None, decode_responses: bool = False, + credential_options: dict | None = None, ) -> "AzureRedisCacheBackend": - """Authenticate via Azure Managed Identity (Entra ID token auth). + """Authenticate via Azure AD (Entra ID token auth). + Resolves an identity through workload identity → managed identity → az CLI. Requires the cache to have *Microsoft Entra Authentication* enabled and the - managed identity to have a Redis data-access role assigned. + identity to have a Redis data-access role assigned. Args: host: e.g. ``.redis.cache.windows.net`` - username: The object ID (or configured Redis username) of the managed identity. + username: The object ID (or configured Redis username) of the identity. port: Redis SSL port (default 6380). db: Database index (default 0). ssl: Enable TLS (default ``True``). client_id: Optional client ID for a user-assigned managed identity. Omit to use the system-assigned identity. + credential_options: Forwarded to ``DefaultAzureCredential`` — see + :mod:`cloudrift.core.azure_credentials`. """ try: - from azure.identity import ManagedIdentityCredential - credential = ManagedIdentityCredential(client_id=client_id) if client_id else ManagedIdentityCredential() + from cloudrift.core.azure_credentials import build_credential + + credential = build_credential(client_id, **(credential_options or {})) provider = _AzureEntraCredentialProvider(credential, username) client = aioredis.Redis( host=host, diff --git a/cloudrift/core/azure_credentials.py b/cloudrift/core/azure_credentials.py new file mode 100644 index 0000000..e93846b --- /dev/null +++ b/cloudrift/core/azure_credentials.py @@ -0,0 +1,80 @@ +"""Shared Azure AD credential construction. + +Every Azure backend authenticates the same way, so the chain is defined once +here instead of being copy-pasted into each provider module. + +The chain that results from the defaults below is:: + + Workload Identity -> Managed Identity -> Azure CLI + +which covers the three environments a Lyzr service actually runs in — AKS with +workload identity, App Service / Container Apps / VM with a managed identity, +and a developer machine with ``az login`` — without any per-environment code. + +Everything excluded below is a developer-machine credential source that is +either ambiguous or actively harmful in a service: + +- ``environment`` — ambient ``AZURE_CLIENT_ID`` / ``AZURE_CLIENT_SECRET`` env + vars would silently shadow the workload's real identity. This is the Azure + counterpart of the SQS ``exclude_env_credentials`` option on + ``AWSSQSBackend.from_iam_role``. +- ``shared_token_cache``, ``visual_studio_code``, ``powershell``, + ``developer_cli`` — stale or user-scoped caches that must never authenticate + a production workload. + +Azure SDK imports are deliberately lazy so a service installing only +``cloudrift[aws]`` never imports ``azure.identity``. + +Note: ``azure-identity`` is pinned ``>=1.15.0``, which predates +``exclude_broker_credential``. It is therefore not set by default; pass it +through ``**overrides`` if you are on a newer release and want it. +""" + +_EXCLUDED_BY_DEFAULT = { + "exclude_environment_credential": True, + "exclude_shared_token_cache_credential": True, + "exclude_visual_studio_code_credential": True, + "exclude_powershell_credential": True, + "exclude_developer_cli_credential": True, +} + + +def _credential_options(client_id: str | None, overrides: dict) -> dict: + """Merge the house defaults, the managed-identity client ID, and caller overrides. + + Overrides are applied last so a caller can always re-enable a source (or + exclude one that is on by default). + """ + options = dict(_EXCLUDED_BY_DEFAULT) + if client_id: + options["managed_identity_client_id"] = client_id + options.update(overrides) + return options + + +def build_async_credential(client_id: str | None = None, **overrides): + """Return an async ``DefaultAzureCredential`` for the standard chain. + + Args: + client_id: Client ID of a *user-assigned* managed identity. Omit to use + the system-assigned identity. + **overrides: Passed straight to ``DefaultAzureCredential``, applied after + the defaults — e.g. ``exclude_cli_credential=True`` to lock a + production service down to managed identity only. + + The caller owns the returned credential and must ``await credential.close()``. + """ + from azure.identity.aio import DefaultAzureCredential + + return DefaultAzureCredential(**_credential_options(client_id, overrides)) + + +def build_credential(client_id: str | None = None, **overrides): + """Return a synchronous ``DefaultAzureCredential`` for the standard chain. + + Sync twin of :func:`build_async_credential`, for the backends whose SDK has + no async client (Redis/Entra, ACS email, MS SQL token provider). + """ + from azure.identity import DefaultAzureCredential + + return DefaultAzureCredential(**_credential_options(client_id, overrides)) diff --git a/cloudrift/crypto/azure_keyvault_keys.py b/cloudrift/crypto/azure_keyvault_keys.py index 7d79007..f7f4a12 100644 --- a/cloudrift/crypto/azure_keyvault_keys.py +++ b/cloudrift/crypto/azure_keyvault_keys.py @@ -4,11 +4,7 @@ ClientAuthenticationError, ResourceNotFoundError, ) -from azure.identity.aio import ( - ClientSecretCredential, - DefaultAzureCredential, - ManagedIdentityCredential, -) +from azure.identity.aio import ClientSecretCredential from azure.keyvault.keys.crypto import EncryptionAlgorithm from azure.keyvault.keys.crypto.aio import CryptographyClient @@ -34,8 +30,7 @@ class AzureKeyVaultKeysBackend(CryptoBackend): Construct via: - ``from_service_principal`` — tenant_id / client_id / client_secret - - ``from_managed_identity`` — managed identity (optionally a client_id), - else DefaultAzureCredential + - ``from_managed_identity`` — workload identity → managed identity → az CLI """ def __init__( @@ -77,14 +72,20 @@ def from_managed_identity( cls, key_id: str, client_id: str | None = None, + credential_options: dict | None = None, **kwargs, ) -> "AzureKeyVaultKeysBackend": - """Authenticate via managed identity (or DefaultAzureCredential).""" - credential = ( - ManagedIdentityCredential(client_id=client_id) - if client_id - else DefaultAzureCredential() - ) + """Authenticate via Azure AD: workload identity → managed identity → az CLI. + + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + (A dict rather than ``**kwargs`` here because ``**kwargs`` already + carries backend options such as ``algorithm``.) + """ + from cloudrift.core.azure_credentials import build_async_credential + + credential = build_async_credential(client_id, **(credential_options or {})) return cls(key_id, credential, **kwargs) # ------------------------------------------------------------------ diff --git a/cloudrift/email/azure_acs.py b/cloudrift/email/azure_acs.py index 0159132..ca77ac9 100644 --- a/cloudrift/email/azure_acs.py +++ b/cloudrift/email/azure_acs.py @@ -65,15 +65,17 @@ def from_managed_identity( endpoint: str, default_from: str | None = None, client_id: str | None = None, + credential_options: dict | None = None, ) -> "AzureACSEmailBackend": - """Authenticate via Azure Managed Identity.""" - from azure.identity import ManagedIdentityCredential + """Authenticate via Azure AD: workload identity → managed identity → az CLI. - credential = ( - ManagedIdentityCredential(client_id=client_id) - if client_id - else ManagedIdentityCredential() - ) + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + """ + from cloudrift.core.azure_credentials import build_credential + + credential = build_credential(client_id, **(credential_options or {})) return cls(endpoint=endpoint, default_from=default_from, credential=credential) @classmethod diff --git a/cloudrift/messaging/__init__.py b/cloudrift/messaging/__init__.py index 2c7df2f..b241195 100644 --- a/cloudrift/messaging/__init__.py +++ b/cloudrift/messaging/__init__.py @@ -2,7 +2,7 @@ Message, MessagingBackend, OutgoingMessage, - send_json, + to_json, ) @@ -54,5 +54,5 @@ def get_queue(provider: str, **kwargs) -> MessagingBackend: "MessagingBackend", "OutgoingMessage", "get_queue", - "send_json", + "to_json", ] diff --git a/cloudrift/messaging/azure_bus.py b/cloudrift/messaging/azure_bus.py index 2709574..72f35cf 100644 --- a/cloudrift/messaging/azure_bus.py +++ b/cloudrift/messaging/azure_bus.py @@ -2,8 +2,13 @@ from azure.core.exceptions import HttpResponseError, ResourceNotFoundError from azure.servicebus import NEXT_AVAILABLE_SESSION, ServiceBusMessage -from azure.servicebus.aio import ServiceBusClient -from azure.servicebus.exceptions import OperationTimeoutError +from azure.servicebus.aio import ServiceBusClient, ServiceBusSender +from azure.servicebus.exceptions import ( + MessagingEntityNotFoundError, + OperationTimeoutError, + ServiceBusConnectionError, + ServiceBusError, +) from cloudrift.core.exceptions import ( FeatureNotSupportedError, @@ -11,15 +16,15 @@ MessagingError, QueueNotFoundError, ) -from cloudrift.messaging.base import Message, MessagingBackend, OutgoingMessage +from cloudrift.messaging.base import Message, MessagingBackend class AzureServiceBusBackend(MessagingBackend): """Azure Service Bus messaging backend (native async via ``azure.servicebus.aio``). - A single ``ServiceBusClient`` (one AMQP connection) is opened lazily and - reused for the lifetime of the backend. Call ``await backend.close()`` - (or use ``async with backend:``) to release the connection. + A single ``ServiceBusClient`` (one AMQP connection) and a single send link + are opened lazily and reused for the lifetime of the backend. Call + ``await backend.close()`` (or use ``async with backend:``) to release them. Use one of the class methods to construct: - ``from_connection_string`` — shared-access connection string @@ -55,6 +60,9 @@ def __init__( self._namespace = fully_qualified_namespace self._credential = credential self._client: ServiceBusClient | None = None + # One long-lived AMQP send link, reused across sends. Opening a sender + # per message costs a full link handshake + teardown on every send. + self._sender: ServiceBusSender | None = None self._lock = asyncio.Lock() # lock_token → (receiver, ServiceBusReceivedMessage) self._pending: dict[str, tuple] = {} @@ -84,15 +92,17 @@ def from_managed_identity( client_id: str | None = None, *, session_enabled: bool = False, + credential_options: dict | None = None, ) -> "AzureServiceBusBackend": - """Authenticate via Azure Managed Identity (system or user-assigned).""" - from azure.identity.aio import ManagedIdentityCredential + """Authenticate via Azure AD: workload identity → managed identity → az CLI. - credential = ( - ManagedIdentityCredential(client_id=client_id) - if client_id - else ManagedIdentityCredential() - ) + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + """ + from cloudrift.core.azure_credentials import build_async_credential + + credential = build_async_credential(client_id, **(credential_options or {})) return cls( queue_name, fully_qualified_namespace=fully_qualified_namespace, @@ -139,6 +149,31 @@ async def _ensure(self) -> ServiceBusClient: self._client = ServiceBusClient(self._namespace, credential=self._credential) return self._client + async def _ensure_sender(self) -> ServiceBusSender: + """Return the cached send link, creating it on first use. + + The sender is deliberately *not* used as an async context manager — + ``__aexit__`` closes it, which is exactly the per-message teardown this + cache exists to avoid. The SDK reopens the underlying AMQP link itself on + retryable errors, so a long-lived sender survives transient drops. + """ + if self._sender is not None: + return self._sender + client = await self._ensure() + async with self._lock: + if self._sender is None: + self._sender = client.get_queue_sender(self.queue_name) + return self._sender + + async def _discard_sender(self) -> None: + """Drop the cached sender so the next send builds a fresh link.""" + sender, self._sender = self._sender, None + if sender is not None: + try: + await sender.close() + except Exception: + pass + async def close(self) -> None: for receiver, _ in list(self._receiver_tokens.values()): try: @@ -147,6 +182,7 @@ async def close(self) -> None: pass self._receiver_tokens.clear() self._pending.clear() + await self._discard_sender() if self._client is not None: await self._client.close() self._client = None @@ -159,7 +195,7 @@ async def close(self) -> None: def _build_message( self, - body: bytes, + body: str, attributes: dict[str, str] | None, group_id: str | None, dedup_id: str | None, @@ -177,57 +213,90 @@ def _build_message( sb_message.message_id = dedup_id return sb_message - async def send( + @staticmethod + def _is_dead_link(exc: Exception) -> bool: + """True if ``exc`` means the cached send link is unusable and must be rebuilt. + + ``ServiceBusConnectionError`` is the link/connection failure. A bare + ``ValueError`` naming a shut-down handler comes from the SDK's + ``_check_live()`` when the sender was already closed — the one state the + SDK's internal retry cannot recover from. + """ + if isinstance(exc, ServiceBusConnectionError): + return True + return isinstance(exc, ValueError) and "shutdown" in str(exc).lower() + + async def _send_with_sender(self, operation): + """Run ``operation(sender)`` on the cached link, rebuilding it once if it is dead.""" + sender = await self._ensure_sender() + try: + return await operation(sender) + except Exception as e: + if not self._is_dead_link(e): + raise + await self._discard_sender() + sender = await self._ensure_sender() + return await operation(sender) + + def _raise_send_error(self, exc: Exception): + # MessagingEntityNotFoundError subclasses ServiceBusError, so it must be + # matched first. ServiceBusError is an AzureError, *not* an + # HttpResponseError — catching only the latter lets AMQP errors escape. + if isinstance(exc, (MessagingEntityNotFoundError, ResourceNotFoundError)): + raise QueueNotFoundError(f"Queue not found: {self.queue_name}") from exc + raise MessageSendError(str(exc)) from exc + + async def _send_json( self, - body: bytes, + body: str, attributes: dict[str, str] | None = None, delay: int = 0, *, group_id: str | None = None, dedup_id: str | None = None, ) -> str: - client = await self._ensure() sb_message = self._build_message(body, attributes, group_id, dedup_id) - try: - async with client.get_queue_sender(self.queue_name) as sender: - if delay: - from datetime import datetime, timedelta, timezone + if delay: + from datetime import datetime, timedelta, timezone - sb_message.scheduled_enqueue_time_utc = datetime.now(timezone.utc) + timedelta( - seconds=delay - ) - await sender.send_messages(sb_message) - return sb_message.message_id or "" - except ResourceNotFoundError as e: - raise QueueNotFoundError(f"Queue not found: {self.queue_name}") from e - except HttpResponseError as e: - raise MessageSendError(str(e)) from e + sb_message.scheduled_enqueue_time_utc = datetime.now(timezone.utc) + timedelta( + seconds=delay + ) - async def send_batch( + async def _op(sender): + await sender.send_messages(sb_message) + return sb_message.message_id or "" + + try: + return await self._send_with_sender(_op) + except (ServiceBusError, HttpResponseError) as e: + self._raise_send_error(e) + + async def _send_json_batch( self, - messages: list[OutgoingMessage], + items: list[tuple[str, dict[str, str] | None]], *, group_id: str | None = None, dedup_ids: list[str] | None = None, ) -> list[str]: - client = await self._ensure() - if dedup_ids is not None and len(dedup_ids) != len(messages): + if dedup_ids is not None and len(dedup_ids) != len(items): raise MessageSendError("dedup_ids must be parallel to messages") sb_messages = [ - self._build_message(m.body, m.attributes, group_id, dedup_ids[i] if dedup_ids else None) - for i, m in enumerate(messages) + self._build_message(body, attributes, group_id, dedup_ids[i] if dedup_ids else None) + for i, (body, attributes) in enumerate(items) ] + + async def _op(sender): + batch = await sender.create_message_batch() + for msg in sb_messages: + batch.add_message(msg) + await sender.send_messages(batch) + return [msg.message_id or "" for msg in sb_messages] + try: - async with client.get_queue_sender(self.queue_name) as sender: - batch = await sender.create_message_batch() - for msg in sb_messages: - batch.add_message(msg) - await sender.send_messages(batch) - return [msg.message_id or "" for msg in sb_messages] - except ResourceNotFoundError as e: - raise QueueNotFoundError(f"Queue not found: {self.queue_name}") from e - except HttpResponseError as e: - raise MessageSendError(str(e)) from e + return await self._send_with_sender(_op) + except (ServiceBusError, HttpResponseError) as e: + self._raise_send_error(e) async def receive( self, diff --git a/cloudrift/messaging/base.py b/cloudrift/messaging/base.py index 60d4435..9fc744f 100644 --- a/cloudrift/messaging/base.py +++ b/cloudrift/messaging/base.py @@ -3,16 +3,34 @@ from dataclasses import dataclass, field +def to_json(payload: dict, *, default=str) -> str: + """Serialize ``payload`` to a JSON string. + + The single serialization point for every backend. ``default`` is passed to + ``json.dumps`` to stringify values it cannot encode natively (``datetime``, + ``Decimal``, ``UUID``, ...). + + Raises: + TypeError: if ``payload`` is not a ``dict``. + """ + if not isinstance(payload, dict): + raise TypeError( + f"send() takes a dict, got {type(payload).__name__}. " + "Pass the object directly — cloudrift serializes to JSON." + ) + return json.dumps(payload, default=default) + + @dataclass class OutgoingMessage: """A message to send via :meth:`MessagingBackend.send_batch`. - ``body`` is the raw payload bytes; ``attributes`` is an optional flat map of - string metadata that maps to SQS ``MessageAttributes`` (String type) and - Service Bus ``application_properties``. + ``body`` is the payload dict — cloudrift serializes it to JSON. + ``attributes`` is an optional flat map of string metadata that maps to SQS + ``MessageAttributes`` (String type) and Service Bus ``application_properties``. """ - body: bytes + body: dict attributes: dict[str, str] | None = None @@ -26,12 +44,13 @@ class Message: dedup_id: str | None = None receive_count: int | None = None - def json(self): - """Decode the raw ``body`` bytes as JSON. + @property + def data(self) -> dict: + """The JSON body decoded to a dict — symmetric with :meth:`MessagingBackend.send`. - Convenience for the common case where the payload was sent with - :func:`send_json`. Raises ``json.JSONDecodeError`` if the body is not - valid JSON. + ``body`` deliberately stays raw bytes so a malformed or non-UTF-8 payload + from a foreign producer is still inspectable for dead-letter triage. + Raises ``json.JSONDecodeError`` if the body is not valid JSON. """ return json.loads(self.body) @@ -39,9 +58,18 @@ def json(self): class MessagingBackend(ABC): """Abstract base class for cloud messaging/queue backends. - The primitive payload is **raw bytes** plus an optional flat ``attributes`` - map (string → string). JSON users should use the :func:`send_json` helper - and :meth:`Message.json` to (de)serialize without touching the byte layer. + The payload primitive is a **dict**. :meth:`send` and :meth:`send_batch` are + concrete here: they serialize through :func:`to_json` and hand the JSON string + to the backend's :meth:`_send_json` / :meth:`_send_json_batch`. A backend never + sees an unserialized payload, and a caller cannot send a non-dict. Read the + payload back with :attr:`Message.data`. + + Subclass authors: implement ``_send_json``/``_send_json_batch``. Never override + ``send``/``send_batch`` — that is what keeps serialization uniform across + providers. + + Optional ``attributes`` is a flat map (string → string) that maps to SQS + ``MessageAttributes`` (String type) / Service Bus ``application_properties``. Backends hold long-lived async clients. Use ``await backend.close()`` (or ``async with backend:``) to release sockets cleanly. @@ -52,25 +80,37 @@ class MessagingBackend(ABC): when the queue has duplicate detection enabled). """ - @abstractmethod + json_default = staticmethod(str) + """``json.dumps`` fallback for non-JSON-native values. Override per backend if + ``Decimal`` fidelity matters.""" + async def send( self, - body: bytes, + payload: dict, attributes: dict[str, str] | None = None, delay: int = 0, *, group_id: str | None = None, dedup_id: str | None = None, ) -> str: - """Send a raw-bytes message with optional attributes. Returns the message ID. + """Serialize ``payload`` to JSON and send it. Returns the message ID. ``attributes`` map to SQS ``MessageAttributes`` (String type) / Service Bus ``application_properties``. group_id/dedup_id apply to FIFO (SQS) or session-enabled (Service Bus) queues. SQS FIFO does not support per-message ``delay``. + + Raises: + TypeError: if ``payload`` is not a ``dict``. """ + return await self._send_json( + to_json(payload, default=self.json_default), + attributes, + delay, + group_id=group_id, + dedup_id=dedup_id, + ) - @abstractmethod async def send_batch( self, messages: list[OutgoingMessage], @@ -83,6 +123,41 @@ async def send_batch( ``group_id`` applies to every message; ``dedup_ids``, if given, must be parallel to ``messages``. """ + return await self._send_json_batch( + [(to_json(m.body, default=self.json_default), m.attributes) for m in messages], + group_id=group_id, + dedup_ids=dedup_ids, + ) + + @abstractmethod + async def _send_json( + self, + body: str, + attributes: dict[str, str] | None, + delay: int, + *, + group_id: str | None, + dedup_id: str | None, + ) -> str: + """Send one already-serialized JSON ``body``. Returns the message ID. + + Backend extension point for :meth:`send` — it has done the serialization + and type checking already. + """ + + @abstractmethod + async def _send_json_batch( + self, + items: list[tuple[str, dict[str, str] | None]], + *, + group_id: str | None, + dedup_ids: list[str] | None, + ) -> list[str]: + """Send already-serialized ``(json_body, attributes)`` pairs. + + Backend extension point for :meth:`send_batch`. ``items`` is parallel to + the caller's ``messages``, so ``dedup_ids`` indexing still lines up. + """ @abstractmethod async def receive( @@ -95,9 +170,10 @@ async def receive( ) -> list[Message]: """Receive messages. wait_time is long-poll duration in seconds. - Each :class:`Message` carries the raw ``body`` bytes and an - ``attributes`` map (string → string) populated from the provider's - message attributes / application properties. + Each :class:`Message` carries the raw ``body`` bytes (use + :attr:`Message.data` for the decoded dict) and an ``attributes`` map + (string → string) populated from the provider's message attributes / + application properties. ``group_id`` receives from a specific session (Service Bus only; SQS cannot filter by group). ``visibility_timeout`` overrides the queue's @@ -151,27 +227,3 @@ async def __aenter__(self) -> "MessagingBackend": async def __aexit__(self, exc_type, exc, tb) -> None: await self.close() - - -async def send_json( - backend: MessagingBackend, - message: dict, - attributes: dict[str, str] | None = None, - delay: int = 0, - *, - group_id: str | None = None, - dedup_id: str | None = None, -) -> str: - """Serialize ``message`` to JSON bytes and send it via ``backend``. - - Backend-agnostic convenience wrapper around :meth:`MessagingBackend.send` - for the common JSON-payload case. Decode the received body with - :meth:`Message.json`. - """ - return await backend.send( - json.dumps(message).encode(), - attributes, - delay, - group_id=group_id, - dedup_id=dedup_id, - ) diff --git a/cloudrift/messaging/sqs.py b/cloudrift/messaging/sqs.py index d7cb764..a61cc84 100644 --- a/cloudrift/messaging/sqs.py +++ b/cloudrift/messaging/sqs.py @@ -11,7 +11,7 @@ MessagingError, QueueNotFoundError, ) -from cloudrift.messaging.base import Message, MessagingBackend, OutgoingMessage +from cloudrift.messaging.base import Message, MessagingBackend class AWSSQSBackend(MessagingBackend): @@ -228,9 +228,9 @@ def _message_attributes(attributes: dict[str, str] | None) -> dict: } } - async def send( + async def _send_json( self, - body: bytes, + body: str, attributes: dict[str, str] | None = None, delay: int = 0, *, @@ -243,28 +243,28 @@ async def send( try: response = await client.send_message( QueueUrl=self.queue_url, - MessageBody=body.decode(), + MessageBody=body, **params, ) return response["MessageId"] except ClientError as e: self._raise(e) - async def send_batch( + async def _send_json_batch( self, - messages: list[OutgoingMessage], + items: list[tuple[str, dict[str, str] | None]], *, group_id: str | None = None, dedup_ids: list[str] | None = None, ) -> list[str]: client = await self._ensure() - if dedup_ids is not None and len(dedup_ids) != len(messages): + if dedup_ids is not None and len(dedup_ids) != len(items): raise MessageSendError("dedup_ids must be parallel to messages") entries = [] - for i, msg in enumerate(messages): + for i, (body, attributes) in enumerate(items): params = self._fifo_params(group_id, dedup_ids[i] if dedup_ids else None) - params.update(self._message_attributes(msg.attributes)) - entries.append({"Id": str(i), "MessageBody": msg.body.decode(), **params}) + params.update(self._message_attributes(attributes)) + entries.append({"Id": str(i), "MessageBody": body, **params}) try: response = await client.send_message_batch(QueueUrl=self.queue_url, Entries=entries) if response.get("Failed"): diff --git a/cloudrift/pubsub/azure_eventgrid.py b/cloudrift/pubsub/azure_eventgrid.py index 209b986..c948370 100644 --- a/cloudrift/pubsub/azure_eventgrid.py +++ b/cloudrift/pubsub/azure_eventgrid.py @@ -44,16 +44,19 @@ def from_managed_identity( cls, endpoint: str, client_id: str | None = None, + credential_options: dict | None = None, ) -> "AzureEventGridBackend": - """Authenticate via Azure Managed Identity.""" - from azure.identity.aio import ManagedIdentityCredential + """Authenticate via Azure AD: workload identity → managed identity → az CLI. + + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + """ from azure.eventgrid.aio import EventGridPublisherClient - credential = ( - ManagedIdentityCredential(client_id=client_id) - if client_id - else ManagedIdentityCredential() - ) + from cloudrift.core.azure_credentials import build_async_credential + + credential = build_async_credential(client_id, **(credential_options or {})) return cls( EventGridPublisherClient(endpoint, credential), credential=credential ) diff --git a/cloudrift/secrets/azure_keyvault.py b/cloudrift/secrets/azure_keyvault.py index 80faa8b..9b9f549 100644 --- a/cloudrift/secrets/azure_keyvault.py +++ b/cloudrift/secrets/azure_keyvault.py @@ -27,16 +27,19 @@ def from_managed_identity( cls, vault_url: str, client_id: str | None = None, + credential_options: dict | None = None, ) -> "AzureKeyVaultBackend": - """Authenticate via Azure Managed Identity (system or user-assigned).""" - from azure.identity.aio import ManagedIdentityCredential + """Authenticate via Azure AD: workload identity → managed identity → az CLI. + + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + """ from azure.keyvault.secrets.aio import SecretClient - credential = ( - ManagedIdentityCredential(client_id=client_id) - if client_id - else ManagedIdentityCredential() - ) + from cloudrift.core.azure_credentials import build_async_credential + + credential = build_async_credential(client_id, **(credential_options or {})) return cls(SecretClient(vault_url=vault_url, credential=credential), credential=credential) @classmethod diff --git a/cloudrift/sql/mssql.py b/cloudrift/sql/mssql.py index d2ec8ce..a34899c 100644 --- a/cloudrift/sql/mssql.py +++ b/cloudrift/sql/mssql.py @@ -174,14 +174,19 @@ def from_entra_managed_identity( port: int | None = None, connection_kwargs: dict | None = None, odbc_driver: str = _DEFAULT_ODBC_DRIVER, + credential_options: dict | None = None, ) -> "MSSQLSQLBackend": - """Authenticate via an Azure managed identity (system- or user-assigned).""" + """Authenticate via Azure AD: workload identity → managed identity → az CLI. + + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + """ def _provider(): - from azure.identity import ManagedIdentityCredential + from cloudrift.core.azure_credentials import build_credential - cred = ManagedIdentityCredential(client_id=client_id) if client_id \ - else ManagedIdentityCredential() + cred = build_credential(client_id, **(credential_options or {})) return cred.get_token(_AAD_TOKEN_SCOPE).token return cls( diff --git a/cloudrift/storage/azure_blob.py b/cloudrift/storage/azure_blob.py index dbad763..4781443 100644 --- a/cloudrift/storage/azure_blob.py +++ b/cloudrift/storage/azure_blob.py @@ -69,15 +69,17 @@ def from_managed_identity( cls, account_url: str, client_id: str | None = None, + credential_options: dict | None = None, ) -> "AzureBlobClient": - """Authenticate via Azure Managed Identity (system or user-assigned).""" - from azure.identity.aio import ManagedIdentityCredential + """Authenticate via Azure AD: workload identity → managed identity → az CLI. - credential = ( - ManagedIdentityCredential(client_id=client_id) - if client_id - else ManagedIdentityCredential() - ) + ``client_id`` selects a user-assigned managed identity; omit it for the + system-assigned one. ``credential_options`` is forwarded to + ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + """ + from cloudrift.core.azure_credentials import build_async_credential + + credential = build_async_credential(client_id, **(credential_options or {})) return cls( BlobServiceClient(account_url, credential=credential), credential=credential, diff --git a/pyproject.toml b/pyproject.toml index b8996b6..c13ddb6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "lyzr-cloudrift" -version = "0.2.6" +version = "0.2.7" description = "Cloud-agnostic abstraction for storage, messaging, document databases, cache, secrets, SQL, crypto (KMS), pub/sub, and email" readme = "README.md" requires-python = ">=3.11" diff --git a/tests/test_azure_credentials.py b/tests/test_azure_credentials.py new file mode 100644 index 0000000..474e9eb --- /dev/null +++ b/tests/test_azure_credentials.py @@ -0,0 +1,203 @@ +"""Tests for the shared Azure AD credential chain. + +These assert against the real ``DefaultAzureCredential`` (constructing one makes +no network calls — the chain is only probed when a token is requested), so they +catch an azure-identity release that renames or drops a kwarg we depend on. +""" + +from unittest.mock import patch + +import pytest + +from cloudrift.core.azure_credentials import ( + _EXCLUDED_BY_DEFAULT, + build_async_credential, + build_credential, +) + + +def _chain(credential): + return [type(c).__name__ for c in credential.credentials] + + +# --------------------------------------------------------------------------- +# The resulting chain +# --------------------------------------------------------------------------- + + +def test_sync_chain_is_workload_then_managed_then_cli(): + chain = _chain(build_credential()) + assert "ManagedIdentityCredential" in chain + assert "AzureCliCredential" in chain + # developer-machine sources are excluded + assert "EnvironmentCredential" not in chain + assert "SharedTokenCacheCredential" not in chain + assert "VisualStudioCodeCredential" not in chain + assert "AzurePowerShellCredential" not in chain + assert "AzureDeveloperCliCredential" not in chain + + +async def test_async_chain_is_workload_then_managed_then_cli(): + credential = build_async_credential() + try: + chain = _chain(credential) + assert "ManagedIdentityCredential" in chain + assert "AzureCliCredential" in chain + assert "EnvironmentCredential" not in chain + assert "SharedTokenCacheCredential" not in chain + finally: + await credential.close() + + +def test_managed_identity_precedes_cli(): + """Production identity must win over a developer's az login on the same box.""" + chain = _chain(build_credential()) + assert chain.index("ManagedIdentityCredential") < chain.index("AzureCliCredential") + + +# --------------------------------------------------------------------------- +# client_id / overrides plumbing +# --------------------------------------------------------------------------- + + +def test_client_id_becomes_managed_identity_client_id(): + with patch("azure.identity.DefaultAzureCredential") as cred_cls: + build_credential("user-assigned-123") + assert cred_cls.call_args.kwargs["managed_identity_client_id"] == "user-assigned-123" + + +def test_no_client_id_omits_the_kwarg_entirely(): + """Passing managed_identity_client_id=None would pin to the system identity oddly.""" + with patch("azure.identity.DefaultAzureCredential") as cred_cls: + build_credential() + assert "managed_identity_client_id" not in cred_cls.call_args.kwargs + + +def test_defaults_are_applied(): + with patch("azure.identity.aio.DefaultAzureCredential") as cred_cls: + build_async_credential() + for key, value in _EXCLUDED_BY_DEFAULT.items(): + assert cred_cls.call_args.kwargs[key] is value + + +def test_overrides_win_over_defaults(): + with patch("azure.identity.DefaultAzureCredential") as cred_cls: + build_credential(exclude_environment_credential=False) + assert cred_cls.call_args.kwargs["exclude_environment_credential"] is False + + +def test_overrides_can_lock_down_to_managed_identity_only(): + chain = _chain(build_credential(exclude_cli_credential=True)) + assert "AzureCliCredential" not in chain + assert "ManagedIdentityCredential" in chain + + +def test_builder_does_not_mutate_the_module_defaults(): + build_credential("a", exclude_environment_credential=False) + assert _EXCLUDED_BY_DEFAULT["exclude_environment_credential"] is True + assert "managed_identity_client_id" not in _EXCLUDED_BY_DEFAULT + + +# --------------------------------------------------------------------------- +# Every Azure backend routes through the shared helper +# --------------------------------------------------------------------------- + + +# (label, provider SDK to skip on, cloudrift module, class, positional args) +ASYNC_FACTORIES = [ + ( + "messaging", + "azure.servicebus", + "cloudrift.messaging.azure_bus", + "AzureServiceBusBackend", + ("ns.servicebus.windows.net", "q"), + ), + ( + "storage", + "azure.storage.blob", + "cloudrift.storage.azure_blob", + "AzureBlobClient", + ("https://acct.blob.core.windows.net",), + ), + ( + "secrets", + "azure.keyvault.secrets", + "cloudrift.secrets.azure_keyvault", + "AzureKeyVaultBackend", + ("https://v.vault.azure.net",), + ), + ( + "pubsub", + "azure.eventgrid", + "cloudrift.pubsub.azure_eventgrid", + "AzureEventGridBackend", + ("https://t.eventgrid.azure.net",), + ), +] + + +@pytest.mark.parametrize( + "label,sdk,module,cls_name,args", ASYNC_FACTORIES, ids=[f[0] for f in ASYNC_FACTORIES] +) +def test_async_backends_use_the_shared_chain(label, sdk, module, cls_name, args): + """Each async Azure backend must build its credential through the shared chain.""" + import importlib + + pytest.importorskip(sdk, reason=f"{label} extra not installed") + backend_cls = getattr(importlib.import_module(module), cls_name) + with patch("azure.identity.aio.DefaultAzureCredential") as cred_cls: + backend_cls.from_managed_identity(*args, "mi-client-id") + kwargs = cred_cls.call_args.kwargs + assert kwargs["managed_identity_client_id"] == "mi-client-id" + assert kwargs["exclude_environment_credential"] is True + + +def test_crypto_backend_uses_the_shared_chain(): + from cloudrift.crypto.azure_keyvault_keys import AzureKeyVaultKeysBackend + + with patch("azure.identity.aio.DefaultAzureCredential") as cred_cls: + AzureKeyVaultKeysBackend.from_managed_identity( + "https://v.vault.azure.net/keys/k", + "mi-client-id", + credential_options={"exclude_cli_credential": True}, + ) + kwargs = cred_cls.call_args.kwargs + assert kwargs["managed_identity_client_id"] == "mi-client-id" + assert kwargs["exclude_cli_credential"] is True + assert kwargs["exclude_environment_credential"] is True + + +def test_email_backend_uses_the_shared_sync_chain(): + from cloudrift.email.azure_acs import AzureACSEmailBackend + + with patch("azure.identity.DefaultAzureCredential") as cred_cls: + AzureACSEmailBackend.from_managed_identity( + "https://acs.communication.azure.com", client_id="mi-client-id" + ) + assert cred_cls.call_args.kwargs["managed_identity_client_id"] == "mi-client-id" + + +def test_mssql_token_provider_uses_the_shared_sync_chain(): + from cloudrift.sql.mssql import MSSQLSQLBackend + + backend = MSSQLSQLBackend.from_entra_managed_identity( + "srv.database.windows.net", "db", client_id="mi-client-id" + ) + with patch("azure.identity.DefaultAzureCredential") as cred_cls: + cred_cls.return_value.get_token.return_value.token = "tok" + assert backend._token_provider() == "tok" + assert cred_cls.call_args.kwargs["managed_identity_client_id"] == "mi-client-id" + + +# --------------------------------------------------------------------------- +# Backend keyword typos must still fail loudly +# --------------------------------------------------------------------------- + + +def test_backend_kwarg_typo_is_not_swallowed_as_a_credential_option(): + """credential_options is an explicit dict, not **kwargs, precisely so that a + misspelled backend option raises instead of silently reaching Azure.""" + from cloudrift.messaging.azure_bus import AzureServiceBusBackend + + with pytest.raises(TypeError, match="session_enable"): + AzureServiceBusBackend.from_managed_identity("ns", "q", session_enable=True) diff --git a/tests/test_messaging.py b/tests/test_messaging.py index 282aaee..623b359 100644 --- a/tests/test_messaging.py +++ b/tests/test_messaging.py @@ -5,7 +5,7 @@ from moto.server import ThreadedMotoServer from cloudrift.core.exceptions import FeatureNotSupportedError, MessageSendError -from cloudrift.messaging import OutgoingMessage, get_queue, send_json +from cloudrift.messaging import OutgoingMessage, get_queue REGION = "us-east-1" QUEUE_NAME = "test-queue" @@ -62,33 +62,47 @@ async def sqs_backend(moto_server, sqs_client): sqs_client.delete_queue(QueueUrl=dlq_url) -async def test_send_and_receive_bytes(sqs_backend): - msg_id = await sqs_backend.send(b"raw payload") +async def test_send_and_receive_dict(sqs_backend): + msg_id = await sqs_backend.send({"raw": "payload"}) assert isinstance(msg_id, str) messages = await sqs_backend.receive(max_messages=1) assert len(messages) == 1 - assert messages[0].body == b"raw payload" + assert messages[0].data == {"raw": "payload"} + # body stays raw bytes so a malformed payload is still inspectable assert isinstance(messages[0].body, bytes) async def test_send_with_attributes_round_trip(sqs_backend): - await sqs_backend.send(b"hi", attributes={"content_type": "text/plain", "source": "unit-test"}) + await sqs_backend.send( + {"greeting": "hi"}, attributes={"content_type": "application/json", "source": "unit-test"} + ) [m] = await sqs_backend.receive(max_messages=1) - assert m.body == b"hi" - assert m.attributes["content_type"] == "text/plain" + assert m.data == {"greeting": "hi"} + assert m.attributes["content_type"] == "application/json" assert m.attributes["source"] == "unit-test" -async def test_send_json_and_message_json_helper(sqs_backend): - msg_id = await send_json(sqs_backend, {"action": "greet", "name": "cloudrift"}) +async def test_send_dict_and_message_data(sqs_backend): + msg_id = await sqs_backend.send({"action": "greet", "name": "cloudrift"}) assert isinstance(msg_id, str) [m] = await sqs_backend.receive(max_messages=1) - assert m.json() == {"action": "greet", "name": "cloudrift"} + assert m.data == {"action": "greet", "name": "cloudrift"} + + +async def test_send_non_ascii_round_trip(sqs_backend): + """The send path is a JSON string, not bytes — non-ASCII must survive intact.""" + await sqs_backend.send({"n": "café", "emoji": "🚀"}) + [m] = await sqs_backend.receive(max_messages=1) + assert m.data == {"n": "café", "emoji": "🚀"} async def test_send_batch(sqs_backend): ids = await sqs_backend.send_batch( - [OutgoingMessage(body=b"1"), OutgoingMessage(body=b"2"), OutgoingMessage(body=b"3")] + [ + OutgoingMessage(body={"n": 1}), + OutgoingMessage(body={"n": 2}), + OutgoingMessage(body={"n": 3}), + ] ) assert len(ids) == 3 @@ -96,25 +110,25 @@ async def test_send_batch(sqs_backend): async def test_send_batch_with_attributes(sqs_backend): await sqs_backend.send_batch( [ - OutgoingMessage(body=b"a", attributes={"k": "1"}), - OutgoingMessage(body=b"b", attributes={"k": "2"}), + OutgoingMessage(body={"v": "a"}, attributes={"k": "1"}), + OutgoingMessage(body={"v": "b"}, attributes={"k": "2"}), ] ) messages = await sqs_backend.receive(max_messages=10) - by_body = {m.body: m.attributes.get("k") for m in messages} - assert by_body == {b"a": "1", b"b": "2"} + by_body = {m.data["v"]: m.attributes.get("k") for m in messages} + assert by_body == {"a": "1", "b": "2"} async def test_delete_message(sqs_backend): - await sqs_backend.send(b"x") + await sqs_backend.send({"x": 1}) messages = await sqs_backend.receive(max_messages=1) assert messages await sqs_backend.delete(messages[0].receipt_handle) async def test_purge(sqs_backend): - await sqs_backend.send(b"a") - await sqs_backend.send(b"b") + await sqs_backend.send({"v": "a"}) + await sqs_backend.send({"v": "b"}) await sqs_backend.purge() messages = await sqs_backend.receive(max_messages=10) assert messages == [] @@ -125,6 +139,135 @@ def test_invalid_provider(): get_queue("rabbitmq", queue_url="x") +# --------------------------------------------------------------------------- +# Serialization contract (cloudrift.messaging.base) +# --------------------------------------------------------------------------- + + +class _SpyBackend: + """Minimal concrete MessagingBackend that records what the ABC hands down.""" + + def __init__(self): + self.sent = [] + self.batches = [] + + async def _send_json(self, body, attributes, delay, *, group_id, dedup_id): + assert isinstance(body, str), f"backend received {type(body).__name__}, not str" + self.sent.append(body) + return "msg-id" + + async def _send_json_batch(self, items, *, group_id, dedup_ids): + self.batches.append(items) + return ["msg-id"] * len(items) + + async def receive(self, max_messages=1, wait_time=0, *, group_id=None, visibility_timeout=None): + return [] + + async def delete(self, receipt_handle): + pass + + async def dead_letter(self, receipt_handle, reason): + pass + + async def get_queue_depth(self): + return 0 + + async def purge(self): + pass + + async def health_check(self): + return True + + +def _spy(): + from cloudrift.messaging import MessagingBackend + + return type("Spy", (_SpyBackend, MessagingBackend), {})() + + +@pytest.mark.parametrize( + "payload,type_name", + [(b"raw", "bytes"), ("str", "str"), ([1, 2], "list"), (None, "NoneType"), (42, "int")], +) +async def test_send_rejects_non_dict(payload, type_name): + with pytest.raises(TypeError, match=f"got {type_name}"): + await _spy().send(payload) + + +async def test_send_batch_rejects_non_dict_body(): + with pytest.raises(TypeError, match="got bytes"): + await _spy().send_batch([OutgoingMessage(body=b"raw")]) + + +async def test_backend_receives_serialized_json_string(): + spy = _spy() + await spy.send({"action": "process", "id": 42}) + assert spy.sent == ['{"action": "process", "id": 42}'] + await spy.send_batch([OutgoingMessage(body={"i": 1}, attributes={"k": "v"})]) + assert spy.batches == [[('{"i": 1}', {"k": "v"})]] + + +async def test_json_default_stringifies_non_native_values(): + """json_default (str) keeps datetime/Decimal/UUID payloads from raising.""" + import uuid + from datetime import datetime, timezone + from decimal import Decimal + + spy = _spy() + await spy.send( + { + "when": datetime(2026, 7, 28, 12, 0, tzinfo=timezone.utc), + "amount": Decimal("10.50"), + "id": uuid.UUID("12345678-1234-5678-1234-567812345678"), + } + ) + payload = json.loads(spy.sent[0]) + assert payload["when"] == "2026-07-28 12:00:00+00:00" + assert payload["amount"] == "10.50" + assert payload["id"] == "12345678-1234-5678-1234-567812345678" + + +def test_message_data_decodes_body(): + from cloudrift.messaging import Message + + m = Message(id="1", body=b'{"a": 1}', receipt_handle="h") + assert m.data == {"a": 1} + assert not hasattr(m, "json"), "Message.json() was removed in favour of .data" + + +def test_backend_must_implement_send_json_hooks(): + """Implementing the old send/send_batch no longer satisfies the ABC.""" + from cloudrift.messaging import MessagingBackend + + class Legacy(MessagingBackend): + async def send(self, *a, **k): + return "x" + + async def send_batch(self, *a, **k): + return [] + + async def receive(self, *a, **k): + return [] + + async def delete(self, receipt_handle): + pass + + async def dead_letter(self, receipt_handle, reason): + pass + + async def get_queue_depth(self): + return 0 + + async def purge(self): + pass + + async def health_check(self): + return True + + with pytest.raises(TypeError, match="_send_json"): + Legacy() + + # --- New tests for P0 features --- @@ -136,20 +279,20 @@ async def test_health_check(sqs_backend): async def test_standard_queue_zero_delay_omits_delay_seconds(sqs_backend): # Regression: DelaySeconds must be omitted when delay == 0 so the same # code path works on FIFO queues (which reject the parameter). - msg_id = await sqs_backend.send(b"ping") + msg_id = await sqs_backend.send({"ping": True}) assert isinstance(msg_id, str) async def test_group_id_on_standard_queue_raises(sqs_backend): with pytest.raises(FeatureNotSupportedError): - await sqs_backend.send(b"x", group_id="g1") + await sqs_backend.send({"x": 1}, group_id="g1") # --- dead_letter / get_queue_depth --- async def test_dead_letter_moves_message_to_dlq(sqs_backend, sqs_client): - await send_json(sqs_backend, {"poison": True, "id": 7}) + await sqs_backend.send({"poison": True, "id": 7}) [m] = await sqs_backend.receive(max_messages=1) await sqs_backend.dead_letter(m.receipt_handle, reason="schema mismatch") @@ -196,7 +339,7 @@ async def test_dead_letter_without_dlq_raises(moto_server, sqs_client): endpoint_url=moto_server, ) try: - await backend.send(b"x") + await backend.send({"x": 1}) [m] = await backend.receive(max_messages=1) with pytest.raises(MessagingError, match="No dead-letter queue configured"): await backend.dead_letter(m.receipt_handle, reason="x") @@ -219,7 +362,7 @@ async def test_dead_letter_with_explicit_dlq_url(moto_server, sqs_client): endpoint_url=moto_server, ) try: - await send_json(backend, {"n": 1}) + await backend.send({"n": 1}) [m] = await backend.receive(max_messages=1) await backend.dead_letter(m.receipt_handle, reason="explicit") resp = sqs_client.receive_message(QueueUrl=dlq_url, MaxNumberOfMessages=1) @@ -231,7 +374,7 @@ async def test_dead_letter_with_explicit_dlq_url(moto_server, sqs_client): async def test_nack_drops_pending_entry(sqs_backend): - await sqs_backend.send(b"retry") + await sqs_backend.send({"retry": True}) [m] = await sqs_backend.receive(max_messages=1) assert m.receipt_handle in sqs_backend._pending await sqs_backend.nack(m.receipt_handle) @@ -243,8 +386,8 @@ async def test_nack_drops_pending_entry(sqs_backend): async def test_get_queue_depth(sqs_backend): assert await sqs_backend.get_queue_depth() == 0 - await sqs_backend.send(b"a") - await sqs_backend.send(b"b") + await sqs_backend.send({"v": "a"}) + await sqs_backend.send({"v": "b"}) assert await sqs_backend.get_queue_depth() == 2 await sqs_backend.purge() @@ -280,20 +423,20 @@ async def fifo_backend(moto_server): async def test_fifo_send_requires_group_id(fifo_backend): with pytest.raises(MessageSendError, match="group_id is required"): - await fifo_backend.send(b"x") + await fifo_backend.send({"x": 1}) async def test_fifo_send_with_delay_raises(fifo_backend): with pytest.raises(FeatureNotSupportedError): - await fifo_backend.send(b"x", delay=5, group_id="g1") + await fifo_backend.send({"x": 1}, delay=5, group_id="g1") async def test_fifo_send_and_receive_exposes_fifo_fields(fifo_backend): - await send_json(fifo_backend, {"n": 1}, group_id="owner-1", dedup_id="d-1") + await fifo_backend.send({"n": 1}, group_id="owner-1", dedup_id="d-1") messages = await fifo_backend.receive(max_messages=1) assert len(messages) == 1 m = messages[0] - assert m.json() == {"n": 1} + assert m.data == {"n": 1} assert m.group_id == "owner-1" assert m.dedup_id == "d-1" assert m.receive_count == 1 @@ -301,8 +444,8 @@ async def test_fifo_send_and_receive_exposes_fifo_fields(fifo_backend): async def test_fifo_dedup_id_suppresses_duplicate(fifo_backend): - await fifo_backend.send(b"1", group_id="g-dedup", dedup_id="same-id") - await fifo_backend.send(b"2", group_id="g-dedup", dedup_id="same-id") + await fifo_backend.send({"n": 1}, group_id="g-dedup", dedup_id="same-id") + await fifo_backend.send({"n": 2}, group_id="g-dedup", dedup_id="same-id") messages = await fifo_backend.receive(max_messages=10) assert len(messages) == 1 await fifo_backend.delete(messages[0].receipt_handle) @@ -310,27 +453,27 @@ async def test_fifo_dedup_id_suppresses_duplicate(fifo_backend): async def test_fifo_ordering_within_group(fifo_backend): for i in range(3): - await send_json(fifo_backend, {"seq": i}, group_id="g-order", dedup_id=f"ord-{i}") + await fifo_backend.send({"seq": i}, group_id="g-order", dedup_id=f"ord-{i}") received = [] while len(received) < 3: messages = await fifo_backend.receive(max_messages=10) if not messages: break for m in messages: - received.append(m.json()["seq"]) + received.append(m.data["seq"]) await fifo_backend.delete(m.receipt_handle) assert received == [0, 1, 2] async def test_fifo_send_batch_with_group_and_dedup_ids(fifo_backend): ids = await fifo_backend.send_batch( - [OutgoingMessage(body=b"1"), OutgoingMessage(body=b"2")], + [OutgoingMessage(body={"n": 1}), OutgoingMessage(body={"n": 2})], group_id="g-batch", dedup_ids=["b-1", "b-2"], ) assert len(ids) == 2 messages = await fifo_backend.receive(max_messages=10) - assert {m.body for m in messages} == {b"1", b"2"} + assert {m.data["n"] for m in messages} == {1, 2} for m in messages: await fifo_backend.delete(m.receipt_handle) @@ -338,18 +481,18 @@ async def test_fifo_send_batch_with_group_and_dedup_ids(fifo_backend): async def test_fifo_send_batch_mismatched_dedup_ids(fifo_backend): with pytest.raises(MessageSendError, match="parallel"): await fifo_backend.send_batch( - [OutgoingMessage(body=b"1")], group_id="g", dedup_ids=["a", "b"] + [OutgoingMessage(body={"n": 1})], group_id="g", dedup_ids=["a", "b"] ) async def test_nack_redelivers_immediately(fifo_backend): - await send_json(fifo_backend, {"task": "retry-me"}, group_id="g-nack", dedup_id="nack-1") + await fifo_backend.send({"task": "retry-me"}, group_id="g-nack", dedup_id="nack-1") first = await fifo_backend.receive(max_messages=1, visibility_timeout=300) assert len(first) == 1 await fifo_backend.nack(first[0].receipt_handle) second = await fifo_backend.receive(max_messages=1) assert len(second) == 1 - assert second[0].json() == {"task": "retry-me"} + assert second[0].data == {"task": "retry-me"} assert second[0].receive_count == 2 await fifo_backend.delete(second[0].receipt_handle) @@ -381,17 +524,15 @@ async def test_dlq_redrive_flow(moto_server): main_q = get_queue("sqs", queue_url=main_url, **creds) dlq = get_queue("sqs", queue_url=dlq_url, **creds) try: - await send_json( - dlq, {"owner_id": "u1", "messages": ["hi"]}, group_id="u1", dedup_id="orig-1" - ) + await dlq.send({"owner_id": "u1", "messages": ["hi"]}, group_id="u1", dedup_id="orig-1") failed = await dlq.receive(max_messages=10) assert len(failed) == 1 m = failed[0] - # Re-send the raw bytes body verbatim to the main queue. - await main_q.send(m.body, group_id=m.group_id, dedup_id="redrive-abc-123") + # Re-send the decoded payload to the main queue with a fresh dedup ID. + await main_q.send(m.data, group_id=m.group_id, dedup_id="redrive-abc-123") await dlq.delete(m.receipt_handle) redriven = await main_q.receive(max_messages=1) - assert redriven[0].json() == {"owner_id": "u1", "messages": ["hi"]} + assert redriven[0].data == {"owner_id": "u1", "messages": ["hi"]} assert redriven[0].group_id == "u1" finally: await main_q.close() @@ -463,15 +604,14 @@ def test_assume_role_omits_external_id_when_absent(): # exclude_env_credentials (prevent ambient env creds shadowing the task role) # --------------------------------------------------------------------------- + def _credential_methods(backend): resolver = backend._session._session.get_component("credential_provider") return [p.METHOD for p in resolver.providers] def test_from_iam_role_keeps_env_provider_by_default(): - backend = get_queue( - "sqs", queue_url="https://sqs.us-east-1.amazonaws.com/123/q", region=REGION - ) + backend = get_queue("sqs", queue_url="https://sqs.us-east-1.amazonaws.com/123/q", region=REGION) assert "env" in _credential_methods(backend) diff --git a/tests/test_messaging_azure.py b/tests/test_messaging_azure.py index e457f36..8abb423 100644 --- a/tests/test_messaging_azure.py +++ b/tests/test_messaging_azure.py @@ -8,9 +8,20 @@ import pytest from azure.servicebus import NEXT_AVAILABLE_SESSION -from azure.servicebus.exceptions import OperationTimeoutError - -from cloudrift.core.exceptions import FeatureNotSupportedError, MessageSendError, MessagingError +from azure.servicebus.exceptions import ( + MessageSizeExceededError, + MessagingEntityNotFoundError, + OperationTimeoutError, + ServiceBusConnectionError, + ServiceBusError, +) + +from cloudrift.core.exceptions import ( + FeatureNotSupportedError, + MessageSendError, + MessagingError, + QueueNotFoundError, +) from cloudrift.messaging.azure_bus import AzureServiceBusBackend from cloudrift.messaging.base import OutgoingMessage @@ -24,15 +35,26 @@ def _make_backend(session_enabled=False): def _mock_sender(): - sender = AsyncMock() - sender.__aenter__.return_value = sender - return sender + # The backend caches the sender and never enters it as a context manager — + # `async with sender` would close the link on every send. + return AsyncMock() def _patch_client(backend, client): backend._client = client +def _sending_backend(session_enabled=False, sender=None): + """Backend wired to a mock client whose get_queue_sender returns `sender`.""" + backend = _make_backend(session_enabled=session_enabled) + client = MagicMock() + client.close = AsyncMock() + sender = sender or _mock_sender() + client.get_queue_sender.return_value = sender + _patch_client(backend, client) + return backend, client, sender + + class _FakeReceivedMessage: """Minimal stand-in for ServiceBusReceivedMessage. @@ -70,61 +92,63 @@ def _make_received_message( async def test_send_sets_session_and_message_id(): - backend = _make_backend(session_enabled=True) - client = MagicMock() - sender = _mock_sender() - client.get_queue_sender.return_value = sender - _patch_client(backend, client) + backend, _, sender = _sending_backend(session_enabled=True) - await backend.send(b'{"n": 1}', group_id="owner-1", dedup_id="d-1") + await backend.send({"n": 1}, group_id="owner-1", dedup_id="d-1") sent = sender.send_messages.call_args[0][0] assert sent.session_id == "owner-1" assert sent.message_id == "d-1" +async def test_send_serializes_dict_to_json_body(): + """The ABC hands the backend a JSON string; the SDK encodes it to the same bytes.""" + backend, _, sender = _sending_backend() + + await backend.send({"n": 1, "s": "café"}) + + sent = sender.send_messages.call_args[0][0] + assert b"".join(sent.body) == b'{"n": 1, "s": "caf\\u00e9"}' + + +async def test_send_rejects_non_dict_payload(): + backend, _, sender = _sending_backend() + with pytest.raises(TypeError, match="got bytes"): + await backend.send(b'{"n": 1}') + sender.send_messages.assert_not_awaited() + + async def test_sessionless_send_to_session_queue_raises(): backend = _make_backend(session_enabled=True) _patch_client(backend, MagicMock()) with pytest.raises(MessageSendError, match="group_id is required"): - await backend.send(b'{"n": 1}') + await backend.send({"n": 1}) async def test_send_without_session_on_plain_queue_ok(): - backend = _make_backend(session_enabled=False) - client = MagicMock() - sender = _mock_sender() - client.get_queue_sender.return_value = sender - _patch_client(backend, client) + backend, _, sender = _sending_backend(session_enabled=False) - await backend.send(b'{"n": 1}') + await backend.send({"n": 1}) sent = sender.send_messages.call_args[0][0] assert sent.session_id is None async def test_send_sets_application_properties_from_attributes(): - backend = _make_backend(session_enabled=False) - client = MagicMock() - sender = _mock_sender() - client.get_queue_sender.return_value = sender - _patch_client(backend, client) + backend, _, sender = _sending_backend(session_enabled=False) - await backend.send(b"raw", attributes={"content_type": "text/plain"}) + await backend.send({"raw": True}, attributes={"content_type": "text/plain"}) sent = sender.send_messages.call_args[0][0] assert sent.application_properties == {"content_type": "text/plain"} async def test_send_batch_sets_per_message_dedup_ids(): - backend = _make_backend(session_enabled=True) - client = MagicMock() sender = _mock_sender() batch = MagicMock() sender.create_message_batch = AsyncMock(return_value=batch) - client.get_queue_sender.return_value = sender - _patch_client(backend, client) + backend, _, _ = _sending_backend(session_enabled=True, sender=sender) ids = await backend.send_batch( - [OutgoingMessage(body=b'{"n": 1}'), OutgoingMessage(body=b'{"n": 2}')], + [OutgoingMessage(body={"n": 1}), OutgoingMessage(body={"n": 2})], group_id="g", dedup_ids=["a", "b"], ) @@ -132,6 +156,125 @@ async def test_send_batch_sets_per_message_dedup_ids(): added = [c.args[0] for c in batch.add_message.call_args_list] assert [m.message_id for m in added] == ["a", "b"] assert all(m.session_id == "g" for m in added) + assert [b"".join(m.body) for m in added] == [b'{"n": 1}', b'{"n": 2}'] + + +async def test_send_batch_mismatched_dedup_ids(): + backend, _, _ = _sending_backend(session_enabled=True) + with pytest.raises(MessageSendError, match="parallel"): + await backend.send_batch( + [OutgoingMessage(body={"n": 1})], group_id="g", dedup_ids=["a", "b"] + ) + + +# --------------------------------------------------------------------------- +# Sender caching — one AMQP send link for the life of the backend +# --------------------------------------------------------------------------- + + +async def test_sender_is_cached_across_sends(): + sender = _mock_sender() + # add_message is sync on the real batch; MagicMock keeps it from returning a coroutine + sender.create_message_batch = AsyncMock(return_value=MagicMock()) + backend, client, _ = _sending_backend(sender=sender) + + await backend.send({"n": 1}) + await backend.send({"n": 2}) + await backend.send_batch([OutgoingMessage(body={"n": 3})]) + + client.get_queue_sender.assert_called_once_with("test-queue") + assert sender.send_messages.await_count == 3 + # the cached link is never closed between sends + sender.close.assert_not_awaited() + + +@pytest.mark.parametrize( + "error", + [ + ServiceBusConnectionError(message="link detached"), + ValueError("The handler has already been shutdown. Please use ServiceBusClient ..."), + ], + ids=["connection_error", "shutdown_handler"], +) +async def test_dead_link_is_rebuilt_and_the_send_retried(error): + sender = _mock_sender() + sender.send_messages.side_effect = [error, None] + backend, client, _ = _sending_backend(sender=sender) + + await backend.send({"n": 1}) + + assert client.get_queue_sender.call_count == 2 + assert sender.send_messages.await_count == 2 + sender.close.assert_awaited_once() # the dead link was discarded + + +async def test_dead_link_failing_twice_surfaces_as_cloudrift_error(): + sender = _mock_sender() + sender.send_messages.side_effect = ServiceBusConnectionError(message="still down") + backend, client, _ = _sending_backend(sender=sender) + + with pytest.raises(MessageSendError, match="still down"): + await backend.send({"n": 1}) + assert client.get_queue_sender.call_count == 2 # rebuilt once, then gave up + + +async def test_non_link_error_does_not_rebuild_the_sender(): + sender = _mock_sender() + sender.send_messages.side_effect = MessageSizeExceededError(message="too big") + backend, client, _ = _sending_backend(sender=sender) + + with pytest.raises(MessageSendError, match="too big"): + await backend.send({"n": 1}) + client.get_queue_sender.assert_called_once() + assert sender.send_messages.await_count == 1 + + +async def test_close_releases_the_cached_sender(): + backend, client, sender = _sending_backend() + await backend.send({"n": 1}) + assert backend._sender is sender + + await backend.close() + + sender.close.assert_awaited_once() + assert backend._sender is None + client.close.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# Error translation — callers only ever see cloudrift exceptions +# --------------------------------------------------------------------------- + + +async def test_service_bus_error_translated_to_message_send_error(): + """ServiceBusError is an AzureError, not an HttpResponseError — it must still translate.""" + sender = _mock_sender() + sender.send_messages.side_effect = ServiceBusError(message="amqp exploded") + backend, _, _ = _sending_backend(sender=sender) + + with pytest.raises(MessageSendError, match="amqp exploded"): + await backend.send({"n": 1}) + + +async def test_entity_not_found_translated_to_queue_not_found(): + sender = _mock_sender() + sender.send_messages.side_effect = MessagingEntityNotFoundError(message="gone") + backend, _, _ = _sending_backend(sender=sender) + + with pytest.raises(QueueNotFoundError, match="test-queue"): + await backend.send({"n": 1}) + + +async def test_batch_overflow_translated_to_message_send_error(): + """batch.add_message raises MessageSizeExceededError once the batch is full.""" + sender = _mock_sender() + batch = MagicMock() + batch.add_message.side_effect = MessageSizeExceededError(message="batch full") + sender.create_message_batch = AsyncMock(return_value=batch) + backend, _, _ = _sending_backend(sender=sender) + + with pytest.raises(MessageSendError, match="batch full"): + await backend.send_batch([OutgoingMessage(body={"n": 1})]) async def test_receive_uses_next_available_session(): @@ -198,7 +341,7 @@ async def test_receive_populates_fifo_fields(): assert m.dedup_id == "d-1" assert m.receive_count == 2 # delivery_count + 1 assert m.body == b'{"n": 1}' - assert m.json() == {"n": 1} + assert m.data == {"n": 1} # application_properties (bytes keys/values) are stringified into attributes. assert m.attributes["content_type"] == "text/plain" @@ -285,7 +428,7 @@ async def test_get_queue_depth_uses_admin_client(): async def test_session_enabled_threads_through_factories(): - with patch("azure.identity.aio.ManagedIdentityCredential"): + with patch("azure.identity.aio.DefaultAzureCredential"): b = AzureServiceBusBackend.from_managed_identity( "ns.servicebus.windows.net", "q", session_enabled=True ) diff --git a/uv.lock b/uv.lock index 2359e14..a576a0e 100644 --- a/uv.lock +++ b/uv.lock @@ -1208,7 +1208,7 @@ wheels = [ [[package]] name = "lyzr-cloudrift" -version = "0.2.6" +version = "0.2.7" source = { editable = "." } [package.optional-dependencies]