normalize io.gcp.DicomSearch

.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "revision": 8
+  "revision": 9
 }

sdks/python/apache_beam/io/gcp/healthcare/dicomclient.py

@@ -65,7 +65,7 @@ def qido_search(
     page_size = 500
 
     if params and 'limit' in params:
-      page_size = params['limit']
+      page_size = int(params['limit'])
     elif params:
       params['limit'] = page_size
     else:
@@ -93,7 +93,8 @@ def qido_search(
           return [], status
       results = response.json()
       output += results
-      if len(results) < page_size:
+      # params values may be str (dict[str,str]); always compare as int.
+      if len(results) < int(page_size):
         # got all the results, return
         break
       offset += len(results)

sdks/python/apache_beam/yaml/standard_io.yaml

@@ -118,6 +118,7 @@
     'ReadFromMongoDB': 'apache_beam.yaml.yaml_io.read_from_mongodb'
     'WriteToMongoDB': 'apache_beam.yaml.yaml_io.write_to_mongodb'
     'ReadFromDelta': 'apache_beam.yaml.yaml_io.read_from_delta'
+    'DicomSearch': 'apache_beam.yaml.yaml_io.dicom_search'
 
 # General File Formats
 # Declared as a renaming transform to avoid exposing all

sdks/python/apache_beam/yaml/yaml_io.py

@@ -48,6 +48,7 @@
 from apache_beam.io.gcp.bigquery import BigQueryDisposition
 from apache_beam.portability.api import schema_pb2
 from apache_beam.typehints import schemas
+from apache_beam.typehints.row_type import RowTypeConstraint
 from apache_beam.utils.timestamp import Timestamp
 from apache_beam.yaml import json_utils
 from apache_beam.yaml import yaml_errors
@@ -909,3 +910,131 @@ def match_all(
           path=str(x.path), size_in_bytes=int(x.size_in_bytes),
           last_updated_in_seconds=float(x.last_updated_in_seconds)
           if x.last_updated_in_seconds is not None else None))
+
+
+_DICOM_SEARCH_OUTPUT_SCHEMA = RowTypeConstraint.from_fields([
+    ('result', str),
+    ('status', str),
+    ('input', str),
+])
+
+
+def _dicom_search_result_to_row(result):
+  if not result.get('success'):
+    raise RuntimeError(
+        'DicomSearch failed with status: %s' % (result.get('status'), ))
+  return beam.Row(
+      result=json.dumps(result.get('result', [])),
+      status=str(result.get('status')),
+      input=json.dumps(result.get('input', {})))
+
+
+def _dicom_search_to_output_row(result):
+  return beam.Row(
+      result=json.dumps(result.get('result', [])),
+      status=str(result.get('status')),
+      input=json.dumps(result.get('input', {})))
+
+
+def _dicom_search_to_error_row(result):
+  inp = result.get('input') or {}
+  if isinstance(inp, Mapping):
+    element = beam.Row(**dict(inp))
+  else:
+    element = inp
+  return beam.Row(
+      element=element,
+      msg='DicomSearch failed with status: %s' % (result.get('status'), ),
+      stack='')
+
+
+@beam.ptransform_fn
+def dicom_search(
+    pcoll, *, buffer_size: int = 8, max_workers: int = 5, error_handling=None):
+  """Searches a Google Cloud Healthcare DICOM store using QIDO-RS.
+
+  This transform takes an input PCollection of Rows describing QIDO search
+  requests and returns Rows with the search results encoded as JSON.
+
+  Each input Row must include:
+
+    - project_id (str): GCP project containing the DICOM store.
+    - region (str): Region where the DICOM store resides.
+    - dataset_id (str): Dataset containing the DICOM store.
+    - dicom_store_id (str): DICOM store id.
+    - search_type (str): One of ``studies``, ``series``, or ``instances``.
+    - params (map of str to str, optional): QIDO search filters.
+
+  Successful outputs are Rows with:
+
+    - result (str): JSON-encoded list of matching DICOM resources.
+    - status (str): HTTP status from the DICOM API.
+    - input (str): JSON-encoded copy of the search request.
+
+  Failed searches raise unless ``error_handling`` is set, in which case they
+  are routed to the configured error output.
+
+  Args:
+    buffer_size: Number of requests to buffer before flushing.
+    max_workers: Maximum number of threads used to issue requests.
+    error_handling: If specified, should be a mapping giving an output into
+      which to emit failed searches, as described at
+      https://beam.apache.org/documentation/sdks/yaml-errors/
+  """
+  try:
+    from apache_beam.io.gcp.healthcare.dicomio import DicomSearch
+  except ImportError as exn:
+    raise ValueError(
+        "GCP dependencies are not installed. Cannot use DicomSearch. "
+        "Please install using 'pip install apache-beam[gcp]'.") from exn
+
+  def row_to_dict(value):
+    if value is None:
+      return None
+    if hasattr(value, '_asdict'):
+      return {k: row_to_dict(v) for k, v in value._asdict().items()}
+    elif hasattr(value, 'as_dict'):
+      return {k: row_to_dict(v) for k, v in value.as_dict().items()}
+    elif isinstance(value, (list, tuple)):
+      return [row_to_dict(v) for v in value]
+    elif isinstance(value, Mapping):
+      return {k: row_to_dict(v) for k, v in value.items()}
+    else:
+      return value
+
+  def normalize_request(value):
+    # YAML Create types params as map[str, str]; qido_search needs int
+    # limit/offset for pagination comparisons.
+    request = row_to_dict(value)
+    params = request.get('params')
+    params = dict(params) if isinstance(params, Mapping) else {}
+    limit = params.get('limit', 500)
+    offset = params.get('offset', 0)
+    params['limit'] = int(limit)
+    params['offset'] = int(offset)
+    request['params'] = params
+    return request
+
+  if error_handling:
+    error_handling = yaml_utils.SafeLineLoader.strip_metadata(error_handling)
+
+  results = (
+      pcoll
+      | beam.Map(normalize_request)
+      | DicomSearch(buffer_size=buffer_size, max_workers=max_workers))
+
+  if error_handling and error_handling.get('output'):
+    return {
+        'good': (
+            results
+            | beam.Filter(lambda r: r.get('success'))
+            | beam.Map(_dicom_search_to_output_row).with_output_types(
+                _DICOM_SEARCH_OUTPUT_SCHEMA)),
+        error_handling['output']: (
+            results
+            | beam.Filter(lambda r: not r.get('success'))
+            | beam.Map(_dicom_search_to_error_row)),
+    }
+
+  return results | beam.Map(_dicom_search_result_to_row).with_output_types(
+      _DICOM_SEARCH_OUTPUT_SCHEMA)

sdks/python/apache_beam/yaml/yaml_io_test.py

@@ -807,6 +807,182 @@ def test_query_and_table_both_raises(self):
           schema={'fields': []})
 
 
+class FakeDicomSearch(beam.PTransform):
+  def __init__(
+      self, buffer_size=8, max_workers=5, client=None, credential=None):
+    self.buffer_size = buffer_size
+    self.max_workers = max_workers
+
+  def expand(self, pcoll):
+    def do_search(element):
+      required = [
+          'project_id',
+          'region',
+          'dataset_id',
+          'dicom_store_id',
+          'search_type',
+      ]
+      for key in required:
+        if key not in element:
+          return {
+              'result': [],
+              'status': 'Must have %s in the dict.' % key,
+              'input': element,
+              'success': False,
+          }
+      if element['search_type'] not in ('instances', 'studies', 'series'):
+        return {
+            'result': [],
+            'status': (
+                'Search type can only be "studies", '
+                '"instances" or "series"'),
+            'input': element,
+            'success': False,
+        }
+      if element.get('project_id') == 'bad_project':
+        return {
+            'result': [],
+            'status': 500,
+            'input': element,
+            'success': False,
+        }
+      params = element.get('params') or {}
+      result = [{'PatientName': 'Alice', 'params': params}]
+      return {
+          'result': result,
+          'status': 200,
+          'input': element,
+          'success': True,
+      }
+
+    return pcoll | beam.Map(do_search)
+
+
+def _patch_dicom_search():
+  """Install a fake dicomio module so tests do not require GCP extras."""
+  import sys
+  import types
+
+  dicomio_mod = types.ModuleType('apache_beam.io.gcp.healthcare.dicomio')
+  dicomio_mod.DicomSearch = FakeDicomSearch
+  return mock.patch.dict(
+      sys.modules, {'apache_beam.io.gcp.healthcare.dicomio': dicomio_mod})
+
+
+class YamlDicomSearchTest(unittest.TestCase):
+  def test_dicom_search_success(self):
+    with _patch_dicom_search():
+      with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+          pickle_library='cloudpickle')) as p:
+        result = (
+            p
+            | beam.Create([
+                beam.Row(
+                    project_id='proj',
+                    region='us-central1',
+                    dataset_id='dataset',
+                    dicom_store_id='store',
+                    search_type='instances',
+                    params={'PatientName': 'Alice'})
+            ])
+            | YamlTransform(
+                '''
+                type: DicomSearch
+                '''))
+        assert_that(
+            result
+            | beam.Map(
+                lambda row:
+                (row.status, json.loads(row.result)[0]['PatientName'])),
+            equal_to([('200', 'Alice')]))
+
+  def test_dicom_search_coerces_string_limit_offset(self):
+    with _patch_dicom_search():
+      with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+          pickle_library='cloudpickle')) as p:
+        result = (
+            p
+            | beam.Create([
+                beam.Row(
+                    project_id='proj',
+                    region='us-central1',
+                    dataset_id='dataset',
+                    dicom_store_id='store',
+                    search_type='instances',
+                    params={
+                        'PatientName': 'Alice', 'limit': '500', 'offset': '0'
+                    })
+            ])
+            | YamlTransform(
+                '''
+                type: DicomSearch
+                '''))
+        assert_that(
+            result
+            | beam.Map(
+                lambda row: (
+                    row.status, json.loads(row.result)[0]['params']['limit'],
+                    json.loads(row.result)[0]['params']['offset'])),
+            equal_to([('200', 500, 0)]))
+
+  def test_dicom_search_with_error_handling(self):
+    with _patch_dicom_search():
+      with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+          pickle_library='cloudpickle')) as p:
+        result = (
+            p
+            | beam.Create([
+                beam.Row(
+                    project_id='proj',
+                    region='us-central1',
+                    dataset_id='dataset',
+                    dicom_store_id='store',
+                    search_type='instances'),
+                beam.Row(
+                    project_id='bad_project',
+                    region='us-central1',
+                    dataset_id='dataset',
+                    dicom_store_id='store',
+                    search_type='instances'),
+            ])
+            | YamlTransform(
+                '''
+                type: DicomSearch
+                config:
+                  error_handling:
+                    output: errors
+                '''))
+        assert_that(
+            result['good'] | beam.Map(lambda row: row.status),
+            equal_to(['200']),
+            label='CheckGood')
+        assert_that(
+            result['errors'] | beam.Map(lambda error: error.msg),
+            equal_to(['DicomSearch failed with status: 500']),
+            label='CheckErrors')
+
+  def test_dicom_search_without_error_handling_raises(self):
+    with self.assertRaises(Exception):
+      with _patch_dicom_search():
+        with beam.Pipeline(
+            options=beam.options.pipeline_options.PipelineOptions(
+                pickle_library='cloudpickle')) as p:
+          _ = (
+              p
+              | beam.Create([
+                  beam.Row(
+                      project_id='bad_project',
+                      region='us-central1',
+                      dataset_id='dataset',
+                      dicom_store_id='store',
+                      search_type='instances')
+              ])
+              | YamlTransform(
+                  '''
+                  type: DicomSearch
+                  '''))
+
+
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)
   unittest.main()