changeset 1262:a737d4cca0d3

python storage sample
author Alain Mazy <am@orthanc.team>
date Tue, 10 Feb 2026 15:18:53 +0100
parents 6775156f8493
children 57976f197f2a
files Sphinx/source/plugins/python.rst Sphinx/source/plugins/python/storage-area-3.py
diffstat 2 files changed, 135 insertions(+), 2 deletions(-) [+]
line wrap: on
line diff
--- a/Sphinx/source/plugins/python.rst	Tue Feb 10 14:56:37 2026 +0100
+++ b/Sphinx/source/plugins/python.rst	Tue Feb 10 15:18:53 2026 +0100
@@ -870,12 +870,21 @@
 <https://en.wikipedia.org/wiki/Object_storage>`__ technology available
 in Python (such as `Ceph
 <https://en.wikipedia.org/wiki/Ceph_(software)>`__ or AWS S3-like
-tools). The performance will not be as good as a C/C++ native plugin
+tools). The performance will likely not be as good as a C/C++ native plugin
 (cf. the :ref:`cloud storage <object-storage>`, the :ref:`PostgreSQL
 <postgresql>` and the :ref:`MySQL <mysql>` plugins), but it can be
 used for prototyping or for basic setups.
 
-Here is a full, self-explaining sample:
+Note that the signatures of these callbacks has changed between
+version 3.3 and 7.1 (not released yet)
+
+Here is a full self-explaining sample for plugin from 7.1
+
+.. literalinclude:: python/storage-area-3.py
+                    :language: python
+
+Here is a full, self-explaining sample for plugin 3.3 and later 
+(still compatible with 7.1 but with reduced features):
 
 .. literalinclude:: python/storage-area.py
                     :language: python
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/Sphinx/source/plugins/python/storage-area-3.py	Tue Feb 10 15:18:53 2026 +0100
@@ -0,0 +1,124 @@
+# Compared to the default Orthanc storage, this storage plugin 
+# uses a flat folder structure but prefixes all file paths with 
+# the PatientID.  This is a sample only and should not be used 
+# in production.
+
+import orthanc
+from typing import Tuple, Optional
+import json
+import os
+import re
+
+orthanc_full_configuration = json.loads(orthanc.GetConfiguration())
+
+root_storage_directory = orthanc_full_configuration.get('StorageDirectory', orthanc_full_configuration.get('IndexDirectory', os.curdir))
+
+orthanc.LogWarning(f"Python Storage: root directory: {root_storage_directory}")
+
+def normalize_filename(s: str, replacement_char: str = "_") -> str:
+    # Replace invalid characters with the replacement character
+    s = re.sub(r'[\\/*?:"<>|]', replacement_char, s)
+    # Replace spaces with the replacement character
+    s = re.sub(r'\s+', replacement_char, s)
+    # Remove leading/trailing replacement characters
+    s = s.strip(replacement_char)
+    # Ensure the string is not empty
+    if not s:
+        s = "undefined"
+    return s
+
+def get_relative_path(patient_id: str, uuid: str, content_type: orthanc.ContentType) -> str:
+    return f"{normalize_filename(patient_id)} - {uuid}{get_extension(content_type)}"
+
+def get_absolute_path(relative_path: str) -> str:
+    return os.path.join(root_storage_directory, relative_path)
+
+def get_extension(content_type: orthanc.ContentType):
+    if content_type == orthanc.ContentType.DICOM:
+        return ".dcm"
+    elif content_type == orthanc.ContentType.DICOM_AS_JSON:
+        return ".json"
+    elif content_type == orthanc.ContentType.DICOM_UNTIL_PIXEL_DATA:
+        return ".dcm.head"
+    else:
+        return ".unk"
+
+def StorageCreate(uuid: str, 
+                  content_type: orthanc.ContentType, 
+                  compression_type: orthanc.CompressionType, 
+                  content: bytes, 
+                  dicom_instance: orthanc.DicomInstance) -> Tuple[orthanc.ErrorCode, Optional[bytes]]:
+    # orthanc.LogInfo("---- StorageCreate")
+    
+    instance_tags = json.loads(dicom_instance.GetInstanceJson())
+    patient_id = instance_tags["0010,0010"]["Value"]
+    try:
+        relative_path = get_relative_path(patient_id, uuid, content_type)
+        path = get_absolute_path(relative_path)
+        with open(path, "wb") as f:
+            f.write(content)
+        
+        # orthanc.LogInfo(f"Successfully created file {path} for {uuid}, wrote {len(content)} bytes")
+        str_custom_data = relative_path
+
+        return orthanc.ErrorCode.SUCCESS, str_custom_data.encode('utf-8')
+    except IOError as e:
+        orthanc.LogError(f"Failed to create storage for {uuid}: {str(e)}")
+        return orthanc.ErrorCode.PLUGIN, None
+    except Exception as e:
+        orthanc.LogError(f"Unexpected error creating storage for {uuid}: {str(e)}")
+        return orthanc.ErrorCode.PLUGIN, None
+
+def StorageReadRange(uuid: str, 
+                     content_type: orthanc.ContentType, 
+                     range_start: int,
+                     size: int,
+                     custom_data: bytes) -> Tuple[orthanc.ErrorCode, Optional[bytes]]:
+    # orthanc.LogInfo("---- StorageReadRange")
+    str_custom_data = custom_data.decode('utf-8')
+    # orthanc.LogInfo(f"---- CustomData: {str_custom_data}")
+    try:
+        relative_path = str_custom_data
+        path = get_absolute_path(relative_path)
+
+        with open(path, "rb") as f:
+            # Move to the start position
+            f.seek(range_start)
+            # Read the specified number of bytes
+            data = f.read(size)
+            
+        # orthanc.LogInfo(f"Read {len(data)} bytes from position {range_start}")
+        return orthanc.ErrorCode.SUCCESS, data
+    except FileNotFoundError:
+        orthanc.LogError(f"File {path} not found for {uuid}")
+        return orthanc.ErrorCode.UNKNOWN_RESOURCE, None
+    except Exception as e:
+        orthanc.LogError(f"Error reading file {path} for {uuid}: {str(e)}")
+        return orthanc.ErrorCode.PLUGIN, None
+
+def StorageRemove(uuid: str,
+                  content_type: orthanc.ContentType,
+                  custom_data: bytes) -> orthanc.ErrorCode:
+    # orthanc.LogInfo("---- StorageRemove")
+    
+    str_custom_data = custom_data.decode('utf-8')
+    # orthanc.LogInfo(f"---- CustomData: {str_custom_data}")
+    try:
+        relative_path = str_custom_data
+        path = get_absolute_path(relative_path)
+        if os.path.exists(path):
+            os.remove(path)
+            # orthanc.LogInfo(f"Successfully removed file {path} for {uuid}")
+        else:
+            orthanc.LogWarning(f"Storage file not found for removal: {path}")
+        
+        return orthanc.ErrorCode.SUCCESS
+    except IOError as e:
+        orthanc.LogError(f"Failed to remove {path} for {uuid}: {str(e)}")
+        return orthanc.ErrorCode.PLUGIN
+    except Exception as e:
+        orthanc.LogError(f"Unexpected error removing {path} for {uuid}: {str(e)}")
+        return orthanc.ErrorCode.PLUGIN
+
+
+orthanc.RegisterStorageArea3(StorageCreate, StorageReadRange, StorageRemove)
\ No newline at end of file