Merge pull request #118 from jaya-krishnan-sr/feat/webclient_support_permissive_uid

feat: add permissive UID validation for DicomWebClient

Author: CPBridge

src/dicomweb_client/uri.py

@@ -629,13 +629,16 @@ def _validate_resource_identifiers_and_suffix(
 
 def _validate_uid(uid: str, permissive: bool) -> None:
     """Validates a DICOM UID."""
-    if len(uid) > _MAX_UID_LENGTH:
-        raise ValueError('UID cannot have more than 64 chars. '
-                         f'Actual count in {uid!r}: {len(uid)}')
-    if not permissive and _REGEX_UID.fullmatch(uid) is None:
-        raise ValueError(f'UID {uid!r} must match regex {_REGEX_UID!r} in '
-                         'conformance with the DICOM Standard.')
-    elif permissive and _REGEX_PERMISSIVE_UID.fullmatch(uid) is None:
+    if not isinstance(uid, str):
+        raise TypeError('DICOM UID must be a string.')
+    if not permissive:
+        if len(uid) > _MAX_UID_LENGTH:
+            raise ValueError('UID cannot have more than 64 chars. '
+                             f'Actual count in {uid!r}: {len(uid)}')
+        if _REGEX_UID.fullmatch(uid) is None:
+            raise ValueError(f'UID {uid!r} must match regex {_REGEX_UID!r} in '
+                             'conformance with the DICOM Standard.')
+    elif _REGEX_PERMISSIVE_UID.fullmatch(uid) is None:
         raise ValueError(f'Permissive mode is enabled. UID {uid!r} must match '
                          f'regex {_REGEX_PERMISSIVE_UID!r}.')
 

src/dicomweb_client/web.py

@@ -32,7 +32,11 @@
 import requests
 import retrying
 
-from dicomweb_client.uri import build_query_string, parse_query_parameters
+from dicomweb_client.uri import (
+    build_query_string,
+    parse_query_parameters,
+    _validate_uid
+)
 
 
 logger = logging.getLogger(__name__)
@@ -200,7 +204,8 @@ def __init__(
         proxies: Optional[Dict[str, str]] = None,
         headers: Optional[Dict[str, str]] = None,
         callback: Optional[Callable] = None,
-        chunk_size: int = 10**6
+        chunk_size: int = 10**6,
+        permissive_uid: bool = False
     ) -> None:
         """Instatiate client.
 
@@ -235,6 +240,14 @@ def __init__(
             when streaming data from the server using chunked transfer encoding
             (used by ``iter_*()`` methods as well as the ``store_instances()``
             method); defaults to ``10**6`` bytes (1MB)
+        permissive_uid: bool, optional
+            If ``True``, relaxes the DICOM Standard validation for UIDs (see
+            main docstring for details). This option is made available since
+            users may be occasionally forced to work with DICOMs or services
+            that may be in violation of the standard. Unless required, use of
+            this flag is **not** recommended, since non-conformant UIDs may
+            lead to unexpected errors downstream, e.g., rejection by a DICOMweb
+            server, etc.
 
         Warning
         -------
@@ -314,6 +327,7 @@ def __init__(
         if callback is not None:
             self._session.hooks = {'response': [callback, ]}
         self._chunk_size = chunk_size
+        self._permissive_uid = permissive_uid
         self.set_http_retry_params()
 
     def _get_transaction_url(self, transaction: _Transaction) -> str:
@@ -2163,29 +2177,6 @@ def delete_study(
             url += f'?{additional_params_query_string}'
         self._http_delete(url)
 
-    def _assert_uid_format(self, uid: str) -> None:
-        """Check whether a DICOM UID has the correct format.
-
-        Parameters
-        ----------
-        uid: str
-            DICOM UID
-
-        Raises
-        ------
-        TypeError
-            When `uid` is not a string
-        ValueError
-            When `uid` doesn't match the regular expression pattern
-            ``"^[.0-9]+$"``
-
-        """
-        if not isinstance(uid, str):
-            raise TypeError('DICOM UID must be a string.')
-        pattern = re.compile('^[.0-9]+$')
-        if not pattern.search(uid):
-            raise ValueError('DICOM UID has invalid format.')
-
     def search_for_series(
         self,
         study_instance_uid: Optional[str] = None,
@@ -2238,7 +2229,7 @@ def search_for_series(
 
         """  # noqa: E501
         if study_instance_uid is not None:
-            self._assert_uid_format(study_instance_uid)
+            _validate_uid(study_instance_uid, self._permissive_uid)
             logger.info(f'search for series of study "{study_instance_uid}"')
         else:
             logger.info('search for series')
@@ -2297,12 +2288,12 @@ def _get_series(
             f'retrieve series "{series_instance_uid}" '
             f'of study "{study_instance_uid}"'
         )
-        self._assert_uid_format(study_instance_uid)
+        _validate_uid(study_instance_uid, self._permissive_uid)
         if series_instance_uid is None:
             raise ValueError(
                 'Series Instance UID is required for retrieval of series.'
             )
-        self._assert_uid_format(series_instance_uid)
+        _validate_uid(series_instance_uid, self._permissive_uid)
         url = self._get_series_url(
             _Transaction.RETRIEVE,
             study_instance_uid,
@@ -2457,7 +2448,7 @@ def retrieve_series_metadata(
                 'Study Instance UID is required for retrieval of '
                 'series metadata.'
             )
-        self._assert_uid_format(study_instance_uid)
+        _validate_uid(study_instance_uid, self._permissive_uid)
         if series_instance_uid is None:
             raise ValueError(
                 'Series Instance UID is required for retrieval of '
@@ -2467,7 +2458,7 @@ def retrieve_series_metadata(
             f'retrieve metadata of series "{series_instance_uid}" '
             f'of study "{study_instance_uid}"'
         )
-        self._assert_uid_format(series_instance_uid)
+        _validate_uid(series_instance_uid, self._permissive_uid)
         url = self._get_series_url(
             _Transaction.RETRIEVE,
             study_instance_uid,
@@ -2660,7 +2651,7 @@ def search_for_instances(
         if series_instance_uid is not None:
             message += f' of series "{series_instance_uid}"'
         if study_instance_uid is not None:
-            self._assert_uid_format(study_instance_uid)
+            _validate_uid(study_instance_uid, self._permissive_uid)
             message += f' of study "{study_instance_uid}"'
         logger.info(message)
         url = self._get_instances_url(
@@ -2727,12 +2718,12 @@ def retrieve_instance(
             raise ValueError(
                 'Study Instance UID is required for retrieval of instance.'
             )
-        self._assert_uid_format(study_instance_uid)
+        _validate_uid(study_instance_uid, self._permissive_uid)
         if series_instance_uid is None:
             raise ValueError(
                 'Series Instance UID is required for retrieval of instance.'
             )
-        self._assert_uid_format(series_instance_uid)
+        _validate_uid(series_instance_uid, self._permissive_uid)
         if sop_instance_uid is None:
             raise ValueError(
                 'SOP Instance UID is required for retrieval of instance.'
@@ -2742,7 +2733,7 @@ def retrieve_instance(
             f'of series "{series_instance_uid}" '
             f'of study "{study_instance_uid}"'
         )
-        self._assert_uid_format(sop_instance_uid)
+        _validate_uid(sop_instance_uid, self._permissive_uid)
         url = self._get_instances_url(
             _Transaction.RETRIEVE,
             study_instance_uid,

tests/test_uri.py

@@ -70,6 +70,21 @@ def test_uid_permissive_invalid(uid):
         URI(_BASE_URL, uid, permissive=True)
 
 
+def test_uid_length_permissive():
+    """Tests that UIDs longer than 64 chars are allowed in permissive mode."""
+    # Create a UID with 65 characters
+    uid_65 = '1' * 65
+
+    # Should fail in strict mode
+    with pytest.raises(ValueError, match='UID cannot have more than 64 chars'):
+        URI(_BASE_URL, uid_65, permissive=False)
+
+    # Should succeed in permissive mode
+    URI(_BASE_URL, uid_65, permissive=True)
+    URI(_BASE_URL, '1.2.3', uid_65, permissive=True)
+    URI(_BASE_URL, '1.2.3', '4.5.6', uid_65, permissive=True)
+
+
 def test_uid_missing_error():
     """Checks *ValueError* is raised when an expected UID is missing."""
     with pytest.raises(ValueError, match='`study_instance_uid` missing with'):

tests/test_web.py

@@ -1608,3 +1608,35 @@ def test_load_xml_response(httpserver, client, cache_dir):
         dataset = _load_xml_dataset(tree)
     assert dataset.RetrieveURL.startswith('https://wadors.hospital.com')
     assert len(dataset.ReferencedSOPSequence) == 2
+
+
+@pytest.mark.parametrize('invalid_uid', [
+    '1.2/3.4', '[email protected]', '1.2.a.4', '1.2.A.4', '.23.4', '1.2..4', '1.2.', '.'
+])
+def test_uid_strict_validation_fail(httpserver, invalid_uid):
+    client = DICOMwebClient(httpserver.url, permissive_uid=False)
+    with pytest.raises(ValueError,
+                       match=f'UID {invalid_uid!r} must match regex'):
+        client.search_for_series(study_instance_uid=invalid_uid)
+
+
+@pytest.mark.parametrize('uid', ['13-abc', 'hello', 'is it', "you're me"])
+def test_uid_permissive_validation_pass(httpserver, uid):
+    client_strict = DICOMwebClient(httpserver.url, permissive_uid=False)
+    client_permissive = DICOMwebClient(httpserver.url, permissive_uid=True)
+
+    # Should fail in strict mode
+    with pytest.raises(ValueError, match=f'UID {uid!r} must match regex'):
+        client_strict.search_for_series(study_instance_uid=uid)
+
+    # Should not raise exception in permissive mode
+    resp = client_permissive.search_for_series(study_instance_uid=uid)
+    assert isinstance(resp, list)
+
+
+@pytest.mark.parametrize('uid', ['1.23.5@', '1/23.4'])
+def test_uid_permissive_validation_fail(httpserver, uid):
+    client = DICOMwebClient(httpserver.url, permissive_uid=True)
+    with pytest.raises(ValueError,
+                       match=f'UID {uid!r} must match regex'):
+        client.search_for_series(study_instance_uid=uid)