mesh-2771:ruff fixes

Author: johnwilliams57-nhs

mesh_client/__init__.py

@@ -16,7 +16,7 @@
 from io import BytesIO
 from itertools import chain
 from types import TracebackType
-from typing import Any, NoReturn, Optional, TypeVar, Union, cast
+from typing import Any, NoReturn, TypeVar, cast
 from urllib.parse import quote as q
 from urllib.parse import urlparse
 
@@ -136,9 +136,9 @@ def optional_header_map() -> dict[str, str]:
 
 # urllib3 'futures' ( not part of 1.26 .. but available in  2.x )
 def reraise(
-    tp: Optional[type[BaseException]],
+    tp: type[BaseException] | None,
     value: BaseException,
-    tb: Optional[TracebackType] = None,
+    tb: TracebackType | None = None,
 ) -> NoReturn:
     try:
         if value.__traceback__ is not tb:
@@ -149,7 +149,7 @@ def reraise(
         tb = None
 
 
-def try_get_endpoint_from_url(url: str) -> Optional[Endpoint]:
+def try_get_endpoint_from_url(url: str) -> Endpoint | None:
     url_parsed = urlparse(url)
     if not url_parsed.hostname:
         return None
@@ -195,7 +195,7 @@ def _looks_like_send_error(status_code: int, response_dict: dict) -> bool:
 
 def _get_send_error_message(
     response_dict: dict,
-) -> tuple[str, Union[SendMessageErrorResponse_v1, SendMessageErrorResponse_v2, dict]]:
+) -> tuple[str, SendMessageErrorResponse_v1 | SendMessageErrorResponse_v2 | dict]:
     if "errorDescription" in response_dict:
         return response_dict["errorDescription"], cast(SendMessageErrorResponse_v1, response_dict)
 
@@ -214,12 +214,12 @@ class MeshRetry(Retry):
 
     def increment(
         self,
-        method: Optional[str] = None,
-        url: Optional[str] = None,
+        method: str | None = None,
+        url: str | None = None,
         response=None,
-        error: Optional[Exception] = None,
-        _pool: Optional[ConnectionPool] = None,
-        _stacktrace: Optional[TracebackType] = None,
+        error: Exception | None = None,
+        _pool: ConnectionPool | None = None,
+        _stacktrace: TracebackType | None = None,
     ) -> "MeshRetry":
         if method != "POST" or not url or not url.endswith("/outbox"):
             return cast(MeshRetry, super().increment(method, url, response, error, _pool, _stacktrace))
@@ -243,11 +243,11 @@ def increment(
 class SSLContextAdapter(HTTPAdapter):
     def __init__(
         self,
-        cert: Optional[Union[tuple[str], tuple[str, str], tuple[str, str, str]]] = None,
-        verify: Optional[Union[str, bool]] = None,
-        check_hostname: Optional[bool] = None,
-        hostname_checks_common_name: Optional[bool] = None,
-        max_retries: Union[int, Retry] = 0,
+        cert: tuple[str] | tuple[str, str] | tuple[str, str, str] | None = None,
+        verify: str | bool | None = None,
+        check_hostname: bool | None = None,
+        hostname_checks_common_name: bool | None = None,
+        max_retries: int | Retry = 0,
     ):
         self.cert = cert
         self.verify = verify
@@ -306,23 +306,23 @@ class MeshClient:
 
     def __init__(  # NOSONAR(S3776) # noqa: C901
         self,  # NOSONAR(S107)
-        url: Union[str, Endpoint],
+        url: str | Endpoint,
         mailbox: str,
         password: str,
-        shared_key: Optional[bytes] = None,
-        cert: Optional[Union[tuple[str], tuple[str, str], tuple[str, str, str]]] = None,
-        verify: Optional[Union[str, bool]] = None,
-        check_hostname: Optional[bool] = None,
-        hostname_checks_common_name: Optional[bool] = None,
+        shared_key: bytes | None = None,
+        cert: tuple[str] | tuple[str, str] | tuple[str, str, str] | None = None,
+        verify: str | bool | None = None,
+        check_hostname: bool | None = None,
+        hostname_checks_common_name: bool | None = None,
         max_chunk_size=75 * 1024 * 1024,
-        proxies: Optional[dict[str, str]] = None,
+        proxies: dict[str, str] | None = None,
         transparent_compress: bool = False,
-        max_retries: Union[int, Retry] = 3,
-        retry_backoff_factor: Union[int, float] = 0.5,
+        max_retries: int | Retry = 3,
+        retry_backoff_factor: int | float = 0.5,
         retry_status_force_list: tuple[int, ...] = (425, 429, 502, 503, 504),
         retry_methods: tuple[str, ...] = ("HEAD", "GET", "PUT", "POST", "DELETE", "OPTIONS", "TRACE"),
-        timeout: Union[int, float] = 10 * 60,
-        application_name: Optional[str] = None,
+        timeout: int | float = 10 * 60,
+        application_name: str | None = None,
     ):
         """
         Create a new MeshClient.
@@ -388,7 +388,7 @@ def __init__(  # NOSONAR(S3776) # noqa: C901
 
         url_lower = self._url.lower()
 
-        self._retries: Union[int, Retry] = 0
+        self._retries: int | Retry = 0
         if isinstance(max_retries, Retry):
             self._retries = max_retries
         elif max_retries:
@@ -498,16 +498,14 @@ def lookup_endpoint(self, ods_code: str, workflow_id: str) -> EndpointLookupResp
         response.raise_for_status()
         return cast(EndpointLookupResponse_v2, response.json())
 
-    def _inbox_v2_page(
-        self, url: Optional[str] = None, params: Optional[dict[str, Any]] = None
-    ) -> ListMessageResponse_v2:
+    def _inbox_v2_page(self, url: str | None = None, params: dict[str, Any] | None = None) -> ListMessageResponse_v2:
         url = url or f"{self.mailbox_url}/inbox"
         response = self._session.get(url, timeout=self._timeout, params=params)
         response.raise_for_status()
 
         return cast(ListMessageResponse_v2, response.json())
 
-    def list_messages(self, max_results: Optional[int] = None, workflow_filter: Optional[str] = None) -> list[str]:
+    def list_messages(self, max_results: int | None = None, workflow_filter: str | None = None) -> list[str]:
         """
             lists messages ids in the inbox; note if workflow_filter is set  it's possible to receive an empty page
             when more results exist outside the first max_results
@@ -521,7 +519,7 @@ def list_messages(self, max_results: Optional[int] = None, workflow_filter: Opti
             list[str]: message ids
         """
 
-        params: dict[str, Union[str, int]] = {}
+        params: dict[str, str | int] = {}
         if max_results:
             if max_results < 10:
                 raise ValueError("if set max_results should be >= 10")
@@ -617,8 +615,8 @@ def send_chunk(
         chunk,
         chunk_num: int,
         total_chunks: int,
-        compress: Optional[bool] = None,
-        message_id: Optional[str] = None,
+        compress: bool | None = None,
+        message_id: str | None = None,
         **kwargs,
     ) -> Response:
         """
@@ -690,8 +688,8 @@ def send_message(
         self,
         recipient: str,
         data,
-        max_chunk_size: Optional[int] = None,
-        compress: Optional[bool] = None,
+        max_chunk_size: int | None = None,
+        compress: bool | None = None,
         **kwargs,
     ) -> str:
         """
@@ -779,7 +777,7 @@ def acknowledge_message(self, message_id: str):
         response.raise_for_status()
 
     def iterate_message_ids(
-        self, workflow_filter: Optional[str] = None, batch_size: Optional[int] = None
+        self, workflow_filter: str | None = None, batch_size: int | None = None
     ) -> Generator[str, None, None]:
         """
             generator lists messages ids in the inbox;
@@ -793,7 +791,7 @@ def iterate_message_ids(
             Generator[str]: message ids
         """
 
-        params: dict[str, Union[int, str]] = {}
+        params: dict[str, int | str] = {}
         if batch_size:
             if batch_size < 10:
                 raise ValueError("if set batch_size should be >= 10")
@@ -802,7 +800,7 @@ def iterate_message_ids(
         if workflow_filter:
             params["workflow_filter"] = workflow_filter
 
-        def _next_messages(page_result: ListMessageResponse_v2) -> tuple[Optional[str], list[str]]:
+        def _next_messages(page_result: ListMessageResponse_v2) -> tuple[str | None, list[str]]:
             return cast(dict[str, str], page_result.get("links", {})).get("next"), cast(
                 list[str], page_result.get("messages", [])
             )
@@ -815,7 +813,7 @@ def _next_messages(page_result: ListMessageResponse_v2) -> tuple[Optional[str],
             next_page, messages = _next_messages(result)
             yield from messages
 
-    def iterate_messages(self, workflow_filter: Optional[str] = None, batch_size: Optional[int] = None):
+    def iterate_messages(self, workflow_filter: str | None = None, batch_size: int | None = None):
         """
             generator lists messages ids in the inbox;
             Iterate over a list of Message objects for all messages in the user's
@@ -886,17 +884,17 @@ class _MessageAttrs:
     message_type: str
     recipient: str
     content_type: str
-    sender: Optional[str] = None
+    sender: str | None = None
 
-    workflow_id: Optional[str] = None
-    filename: Optional[str] = None
-    local_id: Optional[str] = None
-    partner_id: Optional[str] = None
-    chunk_range: Optional[str] = None
+    workflow_id: str | None = None
+    filename: str | None = None
+    local_id: str | None = None
+    partner_id: str | None = None
+    chunk_range: str | None = None
 
-    subject: Optional[str] = None
-    encrypted: Optional[Union[str, bool]] = None
-    compressed: Optional[Union[str, bool]] = None
+    subject: str | None = None
+    encrypted: str | bool | None = None
+    compressed: str | bool | None = None
 
 
 TDefault = TypeVar("TDefault")
@@ -1014,7 +1012,7 @@ def acknowledge(self):
         """
         self._client.acknowledge_message(self._msg_id)
 
-    def mex_header(self, key: str, default: Optional[TDefault] = None) -> Union[str, Optional[TDefault]]:
+    def mex_header(self, key: str, default: TDefault | None = None) -> str | TDefault | None:
         """get a mex header if present
 
         Args:

mesh_client/types.py

@@ -1,11 +1,11 @@
-from typing import Literal, Optional, TypedDict
+from typing import Literal, TypedDict
 
 # https://digital.nhs.uk/developer/api-catalogue/message-exchange-for-social-care-and-health-api#get-/messageexchange/endpointlookup/-ods_code-/-workflow_id-
 
 
 class EndpointLookupItem_v1(TypedDict):
     address: str
-    description: Optional[str]
+    description: str | None
     endpoint_type: Literal["MESH"]
 
 
@@ -16,7 +16,7 @@ class EndpointLookupResponse_v1(TypedDict):
 
 class EndpointLookupItem_v2(TypedDict):
     mailbox_id: str
-    mailbox_name: Optional[str]
+    mailbox_name: str | None
 
 
 class EndpointLookupResponse_v2(TypedDict):
@@ -50,69 +50,69 @@ class AcknowledgeMessageResponse_v1(TypedDict):
 
 # https://digital.nhs.uk/developer/api-catalogue/message-exchange-for-social-care-and-health-api#get-/messageexchange/-mailbox_id-/outbox/tracking/-local_id-
 class TrackingResponse_v1(TypedDict):
-    addressType: Optional[str]
-    checksum: Optional[str]
-    chunkCount: Optional[int]
-    compressFlag: Optional[str]
-    contentEncoding: Optional[str]
-    downloadTimestamp: Optional[str]
+    addressType: str | None
+    checksum: str | None
+    chunkCount: int | None
+    compressFlag: str | None
+    contentEncoding: str | None
+    downloadTimestamp: str | None
     dtsId: str
-    encryptedFlag: Optional[str]
-    expiryTime: Optional[str]
-    failureDate: Optional[str]
-    failureDiagnostic: Optional[str]
-    fileName: Optional[str]
+    encryptedFlag: str | None
+    expiryTime: str | None
+    failureDate: str | None
+    failureDiagnostic: str | None
+    fileName: str | None
     fileSize: int
 
-    isCompressed: Optional[str]
-    linkedMsgId: Optional[str]
-    localId: Optional[str]
-    meshRecipientOdsCode: Optional[str]
+    isCompressed: str | None
+    linkedMsgId: str | None
+    localId: str | None
+    meshRecipientOdsCode: str | None
     messageId: str
-    messageType: Optional[str]
-    partnerId: Optional[str]
+    messageType: str | None
+    partnerId: str | None
 
-    recipient: Optional[str]
-    recipientName: Optional[str]
-    recipientOrgCode: Optional[str]
-    recipientOrgName: Optional[str]
-    recipientSmtp: Optional[str]
+    recipient: str | None
+    recipientName: str | None
+    recipientOrgCode: str | None
+    recipientOrgName: str | None
+    recipientSmtp: str | None
 
-    sender: Optional[str]
-    senderName: Optional[str]
-    senderOdsCode: Optional[str]
-    senderOrgCode: Optional[str]
-    senderOrgName: Optional[str]
-    senderSmtp: Optional[str]
+    sender: str | None
+    senderName: str | None
+    senderOdsCode: str | None
+    senderOrgCode: str | None
+    senderOrgName: str | None
+    senderSmtp: str | None
 
-    status: Optional[str]
+    status: str | None
 
-    statusCode: Optional[str]
-    statusDescription: Optional[str]
+    statusCode: str | None
+    statusDescription: str | None
 
 
 # https://digital.nhs.uk/developer/api-catalogue/message-exchange-for-social-care-and-health-api#get-/messageexchange/-mailbox_id-/outbox/tracking
 class TrackingResponse_v2(TypedDict):
     message_id: str
-    local_id: Optional[str]
-    workflow_id: Optional[str]
-    filename: Optional[str]
+    local_id: str | None
+    workflow_id: str | None
+    filename: str | None
 
-    expiry_time: Optional[str]
-    upload_timestamp: Optional[str]
+    expiry_time: str | None
+    upload_timestamp: str | None
 
-    recipient: Optional[str]
-    recipient_name: Optional[str]
-    recipient_ods_code: Optional[str]
-    recipient_org_code: Optional[str]
-    recipient_org_name: Optional[str]
+    recipient: str | None
+    recipient_name: str | None
+    recipient_ods_code: str | None
+    recipient_org_code: str | None
+    recipient_org_name: str | None
 
-    status_success: Optional[bool]
-    status: Optional[str]
-    status_event: Optional[str]
-    status_timestamp: Optional[str]
-    status_description: Optional[str]
-    status_code: Optional[str]
+    status_success: bool | None
+    status: str | None
+    status_event: str | None
+    status_timestamp: str | None
+    status_description: str | None
+    status_code: str | None
 
 
 # https://digital.nhs.uk/developer/api-catalogue/message-exchange-for-social-care-and-health-api#post-/messageexchange/-mailbox_id-/outbox
@@ -125,13 +125,13 @@ class SendMessageResponse_v2(TypedDict):
 
 
 class SendMessageErrorResponse_v1(TypedDict):
-    messageID: Optional[str]
-    errorEvent: Optional[str]
-    errorCode: Optional[str]
-    errorDescription: Optional[str]
+    messageID: str | None
+    errorEvent: str | None
+    errorCode: str | None
+    errorDescription: str | None
 
 
 class SendMessageErrorResponse_v2(TypedDict):
-    message_id: Optional[str]
-    internal_id: Optional[str]
+    message_id: str | None
+    internal_id: str | None
     detail: list[dict]