Repository: pydicom/pynetdicom

Issue:
Inclusion of Halcyon RT Plan Storage
**Is your feature request related to a problem? Please describe.**
RT plans which come from the halcyon system have an SOP Class UID of 1.2.246.352.70.1.70, rather than the expected 1.2.840.10008.5.1.4.1.1.481.5, and currently this is not supported

**Describe the solution you'd like**
I'd like added support for the identification of this SOP Class UID and convert it to RT Plans.

I've created the solution and am pushing it as a merge request shortly after this. I wanted to make this as an issue file so anyone else can search for it if they haven't pulled the latest version of pynetdicom and have similar issues

Discussion:

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 61d577312..3687a436b 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -2,7 +2,7 @@ version: 2
 jobs:
     build:
         docker:
-            - image: circleci/python:3.10
+            - image: cimg/python:3.10
         steps:
             - checkout
             - run:
diff --git a/docs/changelog/index.rst b/docs/changelog/index.rst
index 523d43290..f043c7358 100644
--- a/docs/changelog/index.rst
+++ b/docs/changelog/index.rst
@@ -7,6 +7,7 @@ Release Notes
 .. toctree::
    :maxdepth: 1
 
+   v2.2.0
    v2.1.0
    v2.0.1
    v2.0.0
diff --git a/docs/changelog/v2.2.0.rst b/docs/changelog/v2.2.0.rst
new file mode 100644
index 000000000..129114ae9
--- /dev/null
+++ b/docs/changelog/v2.2.0.rst
@@ -0,0 +1,18 @@
+.. _v2.2.0:
+
+2.2.0
+=====
+
+Fixes
+.....
+
+
+Enhancements
+............
+
+* Added :func:`~pynetdicom.sop_class.register_uid` to make registering new
+  private and public SOP Classes easier (:issue:`799`)
+
+
+Changes
+.......
diff --git a/docs/index.rst b/docs/index.rst
index 469ad5c67..528953632 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -18,7 +18,8 @@ If you're new to *pynetdicom* then start here:
 
 * **Basics**: :doc:`Installation</tutorials/installation>` |
   :doc:`Writing your first SCU</tutorials/create_scu>` |
-  :doc:`Writing your first SCP</tutorials/create_scp>`
+  :doc:`Writing your first SCP</tutorials/create_scp>` |
+  :doc:`Registering a new SOP Class</tutorials/register_sop_class>`
 
 
 .. _index_guide:
diff --git a/docs/reference/service_classes.rst b/docs/reference/service_classes.rst
index f673a9fe3..2d9588416 100644
--- a/docs/reference/service_classes.rst
+++ b/docs/reference/service_classes.rst
@@ -1,5 +1,7 @@
 .. _api_serviceclasses:
 
+.. py:module:: pynetdicom.service_class
+
 Service Classes (:mod:`pynetdicom.service_class`)
 ==================================================
 
diff --git a/docs/reference/sop_classes.rst b/docs/reference/sop_classes.rst
index 71a70997e..27c17eb6c 100644
--- a/docs/reference/sop_classes.rst
+++ b/docs/reference/sop_classes.rst
@@ -13,6 +13,7 @@ SOP Class Utilities
 .. autosummary::
    :toctree: generated/
 
+   register_uid
    SOPClass
    uid_to_sop_class
    uid_to_service_class
diff --git a/docs/tutorials/index.rst b/docs/tutorials/index.rst
index a195a6294..f62cb86c9 100644
--- a/docs/tutorials/index.rst
+++ b/docs/tutorials/index.rst
@@ -10,3 +10,4 @@ New to *pynetdicom*? Then these tutorials should get you up and running.
    installation
    create_scu
    create_scp
+   register_sop_class
diff --git a/docs/tutorials/register_sop_class.rst b/docs/tutorials/register_sop_class.rst
new file mode 100644
index 000000000..294458461
--- /dev/null
+++ b/docs/tutorials/register_sop_class.rst
@@ -0,0 +1,58 @@
+===========================
+Registering a new SOP Class
+===========================
+
+.. currentmodule:: pynetdicom
+
+You may occasionally come across a private SOP Class UID you like to be able
+to receive, or perhaps there's a public SOP Class from a recent release of the
+DICOM Standard that hasn't yet been added to *pynetdicom*. In this short
+tutorial you'll learn how to register your own UID so it can be used like
+the SOP Classes included by *pynetdicom*.
+
+To register new UIDs you use the :func:`~pynetdicom.sop_class.register_uid` function,
+which takes the UID to be registered, a `keyword` that will be used as the
+variable name for the new UID and the *pynetdicom*
+:mod:`~pynetdicom.service_class` to register the UID with::
+
+    from pynetdicom import register_uid
+    from pynetdicom.service_class import StorageServiceClass
+
+    register_uid(
+        "1.2.246.352.70.1.70",
+        PrivateRTPlanStorage,
+        StorageServiceClass,
+    )
+
+The UID can then be imported from the :mod:`~pynetdicom.sop_class` module and
+used like other UIDs::
+
+    from pynetdicom import AE, evt
+    from pynetdicom.sop_class import PrivateRTPlanStorage
+
+    def handle_store(evt):
+        ds = event.dataset
+        ds.file_meta = event.file_meta
+        ds.save_as(ds.SOPInstanceUID)
+
+        return 0x0000
+
+    ae = AE()
+    # or ae.add_supported_context("1.2.246.352.70.1.70")
+    ae.add_supported_context(PrivateRTPlanStorage)
+    ae.start_server(("localhost", 11112), evt_handlers=[(evt.EVT_C_STORE, handle_store)])
+
+
+When registering a new UID with the
+:class:`~pynetdicom.service_class.QueryRetrieveServiceClass`, you must also
+specify which of the three DIMSE-C message types the UID is to be used with::
+
+    from pynetdicom import register_uid
+    from pynetdicom.service_class import QueryRetrieveServiceClass
+
+    register_uid(
+        "1.2.246.352.70.1.70",
+        PrivateQueryFind,
+        QueryRetrieveServiceClass,
+        dimse_msg_type="C-FIND"  # or "C-GET" or "C-MOVE"
+    )
diff --git a/pynetdicom/__init__.py b/pynetdicom/__init__.py
index 03184c92e..d310af497 100644
--- a/pynetdicom/__init__.py
+++ b/pynetdicom/__init__.py
@@ -63,6 +63,7 @@ from pynetdicom.presentation import (
     UnifiedProcedurePresentationContexts,
     VerificationPresentationContexts,
 )
+from pynetdicom.sop_class import register_uid
 
 
 # Setup default logging
diff --git a/pynetdicom/service_class.py b/pynetdicom/service_class.py
index 5375d0845..c89ef2c58 100644
--- a/pynetdicom/service_class.py
+++ b/pynetdicom/service_class.py
@@ -1538,18 +1538,8 @@ class QueryRetrieveServiceClass(ServiceClass):
         "SpectroscopyData",
         "EncapsulatedDocument",
     ]
-
-    def SCP(self, req: "_QR", context: "PresentationContext") -> None:
-        """The SCP implementation for the Query/Retrieve Service Class.
-
-        Parameters
-        ----------
-        req : dimse_primitives.C_FIND or C_GET or C_MOVE
-            The request primitive received from the peer.
-        context : presentation.PresentationContext
-            The presentation context that the SCP is operating under.
-        """
-        _find_uids = [
+    _SUPPORTED_UIDS = {
+        "C-FIND": [
             "1.2.840.10008.5.1.4.1.2.1.1",
             "1.2.840.10008.5.1.4.1.2.2.1",
             "1.2.840.10008.5.1.4.1.2.3.1",
@@ -1560,8 +1550,8 @@ class QueryRetrieveServiceClass(ServiceClass):
             "1.2.840.10008.5.1.4.44.2",
             "1.2.840.10008.5.1.4.45.2",
             "1.2.840.10008.5.1.4.1.1.200.4",
-        ]
-        _get_uids = [
+        ],
+        "C-GET": [
             "1.2.840.10008.5.1.4.1.2.1.3",
             "1.2.840.10008.5.1.4.1.2.2.3",
             "1.2.840.10008.5.1.4.1.2.3.3",
@@ -1574,8 +1564,8 @@ class QueryRetrieveServiceClass(ServiceClass):
             "1.2.840.10008.5.1.4.44.4",
             "1.2.840.10008.5.1.4.45.4",
             "1.2.840.10008.5.1.4.1.1.200.6",
-        ]
-        _move_uids = [
+        ],
+        "C-MOVE": [
             "1.2.840.10008.5.1.4.1.2.1.2",
             "1.2.840.10008.5.1.4.1.2.2.2",
             "1.2.840.10008.5.1.4.1.2.3.2",
@@ -1587,15 +1577,35 @@ class QueryRetrieveServiceClass(ServiceClass):
             "1.2.840.10008.5.1.4.44.3",
             "1.2.840.10008.5.1.4.45.3",
             "1.2.840.10008.5.1.4.1.1.200.5",
-        ]
+        ],
+    }
 
-        if isinstance(req, C_FIND) and context.abstract_syntax in _find_uids:
+    def SCP(self, req: "_QR", context: "PresentationContext") -> None:
+        """The SCP implementation for the Query/Retrieve Service Class.
+
+        Parameters
+        ----------
+        req : dimse_primitives.C_FIND or C_GET or C_MOVE
+            The request primitive received from the peer.
+        context : presentation.PresentationContext
+            The presentation context that the SCP is operating under.
+        """
+        if (
+            isinstance(req, C_FIND)
+            and context.abstract_syntax in self._SUPPORTED_UIDS["C-FIND"]
+        ):
             self.statuses = QR_FIND_SERVICE_CLASS_STATUS
             self._c_find_scp(req, context)
-        elif isinstance(req, C_GET) and context.abstract_syntax in _get_uids:
+        elif (
+            isinstance(req, C_GET)
+            and context.abstract_syntax in self._SUPPORTED_UIDS["C-GET"]
+        ):
             self.statuses = QR_GET_SERVICE_CLASS_STATUS
             self._get_scp(req, context)
-        elif isinstance(req, C_MOVE) and context.abstract_syntax in _move_uids:
+        elif (
+            isinstance(req, C_MOVE)
+            and context.abstract_syntax in self._SUPPORTED_UIDS["C-MOVE"]
+        ):
             self.statuses = QR_MOVE_SERVICE_CLASS_STATUS
             self._move_scp(req, context)
         else:
@@ -2379,6 +2389,9 @@ class BasicWorklistManagementServiceClass(QueryRetrieveServiceClass):
     """Implementation of the Basic Worklist Management Service Class."""
 
     statuses = QR_FIND_SERVICE_CLASS_STATUS
+    _SUPPORTED_UIDS = {
+        "C-FIND": ["1.2.840.10008.5.1.4.31"],
+    }
 
     def SCP(self, req: "_QR", context: "PresentationContext") -> None:
         """The SCP implementation for Basic Worklist Management.
@@ -2392,7 +2405,7 @@ class BasicWorklistManagementServiceClass(QueryRetrieveServiceClass):
         """
         if (
             isinstance(req, C_FIND)
-            and context.abstract_syntax == "1.2.840.10008.5.1.4.31"
+            and context.abstract_syntax in self._SUPPORTED_UIDS["C-FIND"]
         ):
             self._c_find_scp(req, context)
         else:
@@ -2585,6 +2598,9 @@ class SubstanceAdministrationQueryServiceClass(QueryRetrieveServiceClass):
     """Implementation of the Substance Administration Query Service"""
 
     statuses = SUBSTANCE_ADMINISTRATION_SERVICE_CLASS_STATUS
+    _SUPPORTED_UIDS = {
+        "C-FIND": ["1.2.840.10008.5.1.4.41", "1.2.840.10008.5.1.4.42"],
+    }
 
     def SCP(self, req: "_QR", context: "PresentationContext") -> None:
         """The SCP implementation for the Relevant Patient Information Query
@@ -2597,8 +2613,10 @@ class SubstanceAdministrationQueryServiceClass(QueryRetrieveServiceClass):
         context : presentation.PresentationContext
             The presentation context that the SCP is operating under.
         """
-        uids = ["1.2.840.10008.5.1.4.41", "1.2.840.10008.5.1.4.42"]
-        if isinstance(req, C_FIND) and context.abstract_syntax in uids:
+        if (
+            isinstance(req, C_FIND)
+            and context.abstract_syntax in self._SUPPORTED_UIDS["C-FIND"]
+        ):
             self._c_find_scp(req, context)
         else:
             raise ValueError(
diff --git a/pynetdicom/sop_class.py b/pynetdicom/sop_class.py
index db55d255d..376ec4b02 100644
--- a/pynetdicom/sop_class.py
+++ b/pynetdicom/sop_class.py
@@ -1,6 +1,7 @@
 """Generates the supported SOP Classes and well-known SOP Instances."""
 
 import inspect
+from keyword import iskeyword
 import logging
 import sys
 from typing import Optional, Type, Any, cast, Dict
@@ -56,49 +57,71 @@ def uid_to_service_class(uid: str) -> Type[ServiceClass]:
     """
     if uid in _VERIFICATION_CLASSES.values():
         return VerificationServiceClass
-    elif uid in _QR_CLASSES.values():
+
+    if uid in _QR_CLASSES.values():
         return QueryRetrieveServiceClass
-    elif uid in _STORAGE_CLASSES.values():
+
+    if uid in _STORAGE_CLASSES.values():
         return StorageServiceClass
-    elif uid in _SERVICE_CLASSES:
+
+    if uid in _SERVICE_CLASSES:
         return _SERVICE_CLASSES[uid]
-    elif uid in _APPLICATION_EVENT_CLASSES.values():
+
+    if uid in _APPLICATION_EVENT_CLASSES.values():
         return ApplicationEventLoggingServiceClass
-    elif uid in _BASIC_WORKLIST_CLASSES.values():
+
+    if uid in _BASIC_WORKLIST_CLASSES.values():
         return BasicWorklistManagementServiceClass
-    elif uid in _COLOR_PALETTE_CLASSES.values():
+
+    if uid in _COLOR_PALETTE_CLASSES.values():
         return ColorPaletteQueryRetrieveServiceClass
-    elif uid in _DEFINED_PROCEDURE_CLASSES.values():
+
+    if uid in _DEFINED_PROCEDURE_CLASSES.values():
         return DefinedProcedureProtocolQueryRetrieveServiceClass
-    elif uid in _DISPLAY_SYSTEM_CLASSES.values():
+
+    if uid in _DISPLAY_SYSTEM_CLASSES.values():
         return DisplaySystemManagementServiceClass
-    elif uid in _HANGING_PROTOCOL_CLASSES.values():
+
+    if uid in _HANGING_PROTOCOL_CLASSES.values():
         return HangingProtocolQueryRetrieveServiceClass
-    elif uid in _IMPLANT_TEMPLATE_CLASSES.values():
+
+    if uid in _IMPLANT_TEMPLATE_CLASSES.values():
         return ImplantTemplateQueryRetrieveServiceClass
-    elif uid in _INSTANCE_AVAILABILITY_CLASSES.values():
+
+    if uid in _INSTANCE_AVAILABILITY_CLASSES.values():
         return InstanceAvailabilityNotificationServiceClass
-    elif uid in _MEDIA_CREATION_CLASSES.values():
+
+    if uid in _MEDIA_CREATION_CLASSES.values():
         return MediaCreationManagementServiceClass
-    elif uid in _MEDIA_STORAGE_CLASSES.values():
+
+    if uid in _MEDIA_STORAGE_CLASSES.values():
         return ServiceClass  # Not yet implemented
-    elif uid in _NON_PATIENT_OBJECT_CLASSES.values():
+
+    if uid in _NON_PATIENT_OBJECT_CLASSES.values():
         return NonPatientObjectStorageServiceClass
-    elif uid in _PRINT_MANAGEMENT_CLASSES.values():
+
+    if uid in _PRINT_MANAGEMENT_CLASSES.values():
         return PrintManagementServiceClass
-    elif uid in _PROCEDURE_STEP_CLASSES.values():
+
+    if uid in _PROCEDURE_STEP_CLASSES.values():
         return ProcedureStepServiceClass
-    elif uid in _PROTOCOL_APPROVAL_CLASSES.values():
+
+    if uid in _PROTOCOL_APPROVAL_CLASSES.values():
         return ProtocolApprovalQueryRetrieveServiceClass
-    elif uid in _RELEVANT_PATIENT_QUERY_CLASSES.values():
+
+    if uid in _RELEVANT_PATIENT_QUERY_CLASSES.values():
         return RelevantPatientInformationQueryServiceClass
-    elif uid in _RT_MACHINE_VERIFICATION_CLASSES.values():
+
+    if uid in _RT_MACHINE_VERIFICATION_CLASSES.values():
         return RTMachineVerificationServiceClass
-    elif uid in _STORAGE_COMMITMENT_CLASSES.values():
+
+    if uid in _STORAGE_COMMITMENT_CLASSES.values():
         return StorageCommitmentServiceClass
-    elif uid in _SUBSTANCE_ADMINISTRATION_CLASSES.values():
+
+    if uid in _SUBSTANCE_ADMINISTRATION_CLASSES.values():
         return SubstanceAdministrationQueryServiceClass
-    elif uid in _UNIFIED_PROCEDURE_STEP_CLASSES.values():
+
+    if uid in _UNIFIED_PROCEDURE_STEP_CLASSES.values():
         return UnifiedProcedureStepServiceClass
 
     # No SCP implemented
@@ -112,6 +135,7 @@ class SOPClass(UID):
     """
 
     _service_class: Optional[Type[ServiceClass]] = None
+    _name: str = ""
 
     def __new__(cls: Type["SOPClass"], val: str) -> "SOPClass":
         if isinstance(val, SOPClass):
@@ -429,6 +453,32 @@ _VERIFICATION_CLASSES = {
     "Verification": "1.2.840.10008.1.1",
 }
 
+
+_SERVICE_TO_UID_GROUP = {
+    VerificationServiceClass: _VERIFICATION_CLASSES,
+    QueryRetrieveServiceClass: _QR_CLASSES,
+    StorageServiceClass: _STORAGE_CLASSES,
+    ApplicationEventLoggingServiceClass: _APPLICATION_EVENT_CLASSES,
+    BasicWorklistManagementServiceClass: _BASIC_WORKLIST_CLASSES,
+    ColorPaletteQueryRetrieveServiceClass: _COLOR_PALETTE_CLASSES,
+    DefinedProcedureProtocolQueryRetrieveServiceClass: _DEFINED_PROCEDURE_CLASSES,
+    DisplaySystemManagementServiceClass: _DISPLAY_SYSTEM_CLASSES,
+    HangingProtocolQueryRetrieveServiceClass: _HANGING_PROTOCOL_CLASSES,
+    ImplantTemplateQueryRetrieveServiceClass: _IMPLANT_TEMPLATE_CLASSES,
+    InstanceAvailabilityNotificationServiceClass: _INSTANCE_AVAILABILITY_CLASSES,
+    MediaCreationManagementServiceClass: _MEDIA_CREATION_CLASSES,
+    NonPatientObjectStorageServiceClass: _NON_PATIENT_OBJECT_CLASSES,
+    PrintManagementServiceClass: _PRINT_MANAGEMENT_CLASSES,
+    ProcedureStepServiceClass: _PROCEDURE_STEP_CLASSES,
+    ProtocolApprovalQueryRetrieveServiceClass: _PROTOCOL_APPROVAL_CLASSES,
+    RelevantPatientInformationQueryServiceClass: _RELEVANT_PATIENT_QUERY_CLASSES,
+    RTMachineVerificationServiceClass: _RT_MACHINE_VERIFICATION_CLASSES,
+    StorageCommitmentServiceClass: _STORAGE_COMMITMENT_CLASSES,
+    SubstanceAdministrationQueryServiceClass: _SUBSTANCE_ADMINISTRATION_CLASSES,
+    UnifiedProcedureStepServiceClass: _UNIFIED_PROCEDURE_STEP_CLASSES,
+}
+
+
 # pylint: enable=line-too-long
 _generate_sop_classes(_APPLICATION_EVENT_CLASSES)
 _generate_sop_classes(_BASIC_WORKLIST_CLASSES)
@@ -488,6 +538,87 @@ def uid_to_sop_class(uid: str) -> SOPClass:
     return sop_class
 
 
+def register_uid(
+    uid: str,
+    keyword: str,
+    service_class: Type[ServiceClass],
+    dimse_msg_type: str = "",
+) -> None:
+    """Register a private or public SOP Class UID `uid` with the
+    :mod:`~pynetdicom.sop_class` module.
+
+    Examples
+    --------
+
+    Register the UID ``1.2.246.352.70.1.70`` with the
+    :class:`~pynetdicom.service_class.StorageServiceClass`::
+
+        >>> from pynetdicom import register_uid
+        >>> from pynetdicom.service_class import StorageServiceClass
+        >>> register_uid(
+        ...     "1.2.246.352.70.1.70",
+        ...     "FooStorage",
+        ...     StorageServiceClass,
+        ... )
+
+    Using a UID after registration::
+
+        >>> from pynetdicom import AE
+        >>> from pynetdicom.sop_class import FooStorage
+        >>> ae = AE()
+        >>> ae.add_supported_context(FooStorage)
+
+    Parameters
+    ----------
+    uid : str
+        The UID to be registered.
+    keyword : str
+        The keyword to use for the UID, must be a valid Python identifier and
+        may not be a Python keyword.
+    service_class : pynetdicom.service_class.ServiceClass
+        The service that the `uid` will be registered with, such as
+        :class:`~pynetdicom.service_class.StorageServiceClass`. Note that this
+        must be the class object itself and not a class instance.
+    dimse_msg_type : str, optional
+        If `service_class` is
+        :class:`~pynetdicom.service_class.QueryRetrieveServiceClass` then this
+        should be the DIMSE service message type that the `uid` is being
+        registered to. One of (``"C-FIND"``, ``"C-GET"``, ``"C-MOVE"``).
+    """
+    if not keyword.isidentifier() or iskeyword(keyword):
+        raise ValueError(
+            f"The keyword '{keyword}' is not a valid Python identifier or is "
+            "a Python keyword"
+        )
+
+    if not inspect.isclass(service_class):
+        raise TypeError("'service_class' must be a class object not a class instance")
+
+    if not issubclass(service_class, ServiceClass):
+        raise TypeError(
+            "'service_class' must be a ServiceClass subclass object "
+            "such as 'StorageServiceClass'"
+        )
+
+    group = _SERVICE_TO_UID_GROUP[service_class]
+    group[keyword] = uid
+
+    sop_class = SOPClass(uid)
+    sop_class._service_class = uid_to_service_class(uid)
+    globals()[keyword] = sop_class
+
+    if issubclass(service_class, QueryRetrieveServiceClass):
+        if service_class is QueryRetrieveServiceClass:
+            if dimse_msg_type not in ("C-FIND", "C-GET", "C-MOVE"):
+                raise ValueError(
+                    "'dimse_msg_type' must be 'C-FIND', 'C-GET' or 'C-MOVE' "
+                    "when registering a UID with QueryRetrieveServiceClass"
+                )
+            service_class._SUPPORTED_UIDS[dimse_msg_type].append(uid)
+        else:
+            service_class._SUPPORTED_UIDS["C-FIND"].append(uid)
+
+
 # Well-known SOP Instance UIDs for the supported Service Classes
 DisplaySystemInstance = UID("1.2.840.10008.5.1.1.40.1")
 """``1.2.840.10008.5.1.1.40.1``


Test patch:
diff --git a/pynetdicom/tests/test_service_qr.py b/pynetdicom/tests/test_service_qr.py
index 4df2a067d..6cbf82ebc 100644
--- a/pynetdicom/tests/test_service_qr.py
+++ b/pynetdicom/tests/test_service_qr.py
@@ -26,6 +26,8 @@ from pynetdicom import (
     evt,
     build_role,
     debug_logger,
+    sop_class,
+    register_uid,
 )
 from pynetdicom.dimse_primitives import C_FIND, C_GET, C_MOVE, C_STORE
 from pynetdicom.presentation import PresentationContext
@@ -40,6 +42,8 @@ from pynetdicom.sop_class import (
     PatientRootQueryRetrieveInformationModelGet,
     PatientRootQueryRetrieveInformationModelMove,
     CompositeInstanceRetrieveWithoutBulkDataGet,
+    _QR_CLASSES,
+    _BASIC_WORKLIST_CLASSES,
 )
 
 
@@ -60,6 +64,20 @@ def test_unknown_sop_class():
         service.SCP(None, context)
 
 
+@pytest.fixture()
+def register_new_uid_find():
+    register_uid(
+        "1.2.3.4",
+        "NewFind",
+        QueryRetrieveServiceClass,
+        "C-FIND",
+    )
+    yield
+    del _QR_CLASSES["NewFind"]
+    delattr(sop_class, "NewFind")
+    QueryRetrieveServiceClass._SUPPORTED_UIDS["C-FIND"].remove("1.2.3.4")
+
+
 class TestQRFindServiceClass:
     """Test the QueryRetrieveFindServiceClass"""
 
@@ -1175,6 +1193,54 @@ class TestQRFindServiceClass:
         assert assoc.is_aborted
         scp.shutdown()
 
+    def test_register(self, register_new_uid_find):
+        """Test registering a new UID"""
+        from pynetdicom.sop_class import NewFind
+
+        def handle(event):
+            yield 0xFF01, self.query
+            yield 0x0000, None
+            yield 0xA700, None
+
+        handlers = [(evt.EVT_C_FIND, handle)]
+
+        self.ae = ae = AE()
+        ae.add_supported_context(NewFind)
+        ae.add_requested_context(NewFind, ExplicitVRLittleEndian)
+        scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
+
+        ae.acse_timeout = 5
+        ae.dimse_timeout = 5
+        assoc = ae.associate("localhost", 11112)
+        assert assoc.is_established
+        result = assoc.send_c_find(self.query, NewFind)
+        status, identifier = next(result)
+        assert status.Status == 0xFF01
+        assert identifier == self.query
+        status, identifier = next(result)
+        assert status.Status == 0x0000
+        assert identifier is None
+        with pytest.raises(StopIteration):
+            next(result)
+
+        assoc.release()
+        assert assoc.is_released
+        scp.shutdown()
+
+
+@pytest.fixture()
+def register_new_uid_get():
+    register_uid(
+        "1.2.3.4",
+        "NewGet",
+        QueryRetrieveServiceClass,
+        "C-GET",
+    )
+    yield
+    del _QR_CLASSES["NewGet"]
+    delattr(sop_class, "NewGet")
+    QueryRetrieveServiceClass._SUPPORTED_UIDS["C-GET"].remove("1.2.3.4")
+
 
 class TestQRGetServiceClass:
     def setup_method(self):
@@ -3501,6 +3567,63 @@ class TestQRGetServiceClass:
         assert assoc.is_released
         scp.shutdown()
 
+    def test_register(self, register_new_uid_get):
+        """Test registering a new UID"""
+        from pynetdicom.sop_class import NewGet
+
+        def handle(event):
+            yield 1
+            yield 0xFF00, self.ds
+
+        def handle_store(event):
+            return 0x0000
+
+        handlers = [(evt.EVT_C_GET, handle)]
+
+        self.ae = ae = AE()
+        ae.add_supported_context(NewGet)
+        ae.add_supported_context(CTImageStorage, scu_role=False, scp_role=True)
+        ae.add_requested_context(NewGet)
+        ae.add_requested_context(CTImageStorage)
+        scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
+
+        role = build_role(CTImageStorage, scp_role=True)
+        handlers = [(evt.EVT_C_STORE, handle_store)]
+
+        ae.acse_timeout = 5
+        ae.dimse_timeout = 5
+        assoc = ae.associate("localhost", 11112, ext_neg=[role], evt_handlers=handlers)
+        assert assoc.is_established
+        result = assoc.send_c_get(self.query, NewGet)
+        status, identifier = next(result)
+        assert status.Status == 0xFF00
+        assert identifier is None
+        status, identifier = next(result)
+        assert status.Status == 0x0000
+        assert status.NumberOfFailedSuboperations == 0
+        assert status.NumberOfWarningSuboperations == 0
+        assert status.NumberOfCompletedSuboperations == 1
+        assert identifier is None
+        pytest.raises(StopIteration, next, result)
+
+        assoc.release()
+        assert assoc.is_released
+        scp.shutdown()
+
+
+@pytest.fixture()
+def register_new_uid_move():
+    register_uid(
+        "1.2.3.4",
+        "NewMove",
+        QueryRetrieveServiceClass,
+        "C-MOVE",
+    )
+    yield
+    del _QR_CLASSES["NewMove"]
+    delattr(sop_class, "NewMove")
+    QueryRetrieveServiceClass._SUPPORTED_UIDS["C-MOVE"].remove("1.2.3.4")
+
 
 class TestQRMoveServiceClass:
     def setup_method(self):
@@ -5966,6 +6089,46 @@ class TestQRMoveServiceClass:
         assoc.release()
         scp.shutdown()
 
+    def test_register(self, register_new_uid_move):
+        """Test registering a new UID"""
+        from pynetdicom.sop_class import NewMove
+
+        def handle(event):
+            yield self.destination
+            yield 1
+            yield 0xFF00, self.ds
+
+        def handle_store(event):
+            return 0x0000
+
+        handlers = [(evt.EVT_C_MOVE, handle), (evt.EVT_C_STORE, handle_store)]
+
+        self.ae = ae = AE()
+        ae.add_supported_context(NewMove)
+        ae.add_supported_context(CTImageStorage, scu_role=False, scp_role=True)
+        ae.add_requested_context(NewMove)
+        ae.add_requested_context(CTImageStorage)
+        scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
+
+        ae.acse_timeout = 5
+        ae.dimse_timeout = 5
+        assoc = ae.associate("localhost", 11112)
+        assert assoc.is_established
+        result = assoc.send_c_move(self.query, "TESTMOVE", NewMove)
+        status, identifier = next(result)
+        assert status.Status == 0xFF00
+        assert identifier is None
+        status, identifier = next(result)
+        assert status.Status == 0x0000
+        assert status.NumberOfFailedSuboperations == 0
+        assert status.NumberOfWarningSuboperations == 0
+        assert status.NumberOfCompletedSuboperations == 1
+        assert identifier is None
+        pytest.raises(StopIteration, next, result)
+
+        assoc.release()
+        scp.shutdown()
+
 
 class TestQRCompositeInstanceWithoutBulk:
     """Tests for QR + Composite Instance Without Bulk Data"""
@@ -6174,6 +6337,20 @@ class TestQRCompositeInstanceWithoutBulk:
         scp.shutdown()
 
 
+@pytest.fixture()
+def register_new_uid_bwm():
+    register_uid(
+        "1.2.3.4",
+        "NewFind",
+        BasicWorklistManagementServiceClass,
+        "C-FIND",
+    )
+    yield
+    del _BASIC_WORKLIST_CLASSES["NewFind"]
+    delattr(sop_class, "NewFind")
+    BasicWorklistManagementServiceClass._SUPPORTED_UIDS["C-FIND"].remove("1.2.3.4")
+
+
 class TestBasicWorklistServiceClass:
     """Tests for BasicWorklistManagementServiceClass."""
 
@@ -6198,8 +6375,9 @@ class TestBasicWorklistServiceClass:
             context = build_context("1.2.3.4")
             bwm.SCP(None, context)
 
-    def test_pending_success(self):
-        """Test handler yielding pending then success status"""
+    def test_register(self, register_new_uid_bwm):
+        """Test registering a new UID"""
+        from pynetdicom.sop_class import NewFind
 
         def handle(event):
             yield 0xFF01, self.query
@@ -6209,15 +6387,15 @@ class TestBasicWorklistServiceClass:
         handlers = [(evt.EVT_C_FIND, handle)]
 
         self.ae = ae = AE()
-        ae.add_supported_context(ModalityWorklistInformationFind)
-        ae.add_requested_context(ModalityWorklistInformationFind)
+        ae.add_supported_context(NewFind)
+        ae.add_requested_context(NewFind)
         scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
 
         ae.acse_timeout = 5
         ae.dimse_timeout = 5
         assoc = ae.associate("localhost", 11112)
         assert assoc.is_established
-        result = assoc.send_c_find(self.query, ModalityWorklistInformationFind)
+        result = assoc.send_c_find(self.query, NewFind)
         status, identifier = next(result)
         assert status.Status == 0xFF01
         assert identifier == self.query
diff --git a/pynetdicom/tests/test_service_storage.py b/pynetdicom/tests/test_service_storage.py
index 2c3b5d2d1..ff5e229cf 100644
--- a/pynetdicom/tests/test_service_storage.py
+++ b/pynetdicom/tests/test_service_storage.py
@@ -1,5 +1,6 @@
 """Tests for the StorageServiceClass."""
 
+from copy import deepcopy
 from io import BytesIO
 import os
 from pathlib import Path
@@ -11,13 +12,15 @@ from pydicom import dcmread
 from pydicom.dataset import Dataset, FileMetaDataset
 from pydicom.uid import ExplicitVRLittleEndian, ImplicitVRLittleEndian
 
-from pynetdicom import AE, _config, evt, debug_logger
+from pynetdicom import AE, _config, evt, debug_logger, register_uid, sop_class
 from pynetdicom.dimse_primitives import C_STORE
 from pynetdicom.pdu_primitives import SOPClassExtendedNegotiation
 from pynetdicom.sop_class import (
     Verification,
     CTImageStorage,
+    _STORAGE_CLASSES,
 )
+from pynetdicom.service_class import StorageServiceClass
 
 try:
     from pynetdicom.status import Status
@@ -41,6 +44,18 @@ def enable_unrestricted():
     _config.UNRESTRICTED_STORAGE_SERVICE = False
 
 
+@pytest.fixture()
+def register_new_uid():
+    register_uid(
+        "1.2.3.4",
+        "NewStorage",
+        StorageServiceClass,
+    )
+    yield
+    del _STORAGE_CLASSES["NewStorage"]
+    delattr(sop_class, "NewStorage")
+
+
 class TestStorageServiceClass:
     """Test the StorageServiceClass"""
 
@@ -799,3 +814,35 @@ class TestStorageServiceClass:
         assert recv == ["CompressedSamples^CT1", "Private", "Unknown^Public"]
 
         scp.shutdown()
+
+    def test_register(self, register_new_uid):
+        """Test registering a new UID."""
+        from pynetdicom.sop_class import NewStorage
+
+        attrs = {}
+
+        def handle(event):
+            attrs["uid"] = event.dataset.SOPClassUID
+            return 0x0000
+
+        ds = deepcopy(DATASET)
+        ds.SOPClassUID = NewStorage
+
+        handlers = [(evt.EVT_C_STORE, handle)]
+
+        self.ae = ae = AE()
+        ae.add_supported_context(NewStorage)
+        ae.add_requested_context(NewStorage)
+        scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
+
+        assoc = ae.associate("localhost", 11112)
+        assert assoc.is_established
+        rsp = assoc.send_c_store(ds)
+        assert rsp.Status == 0x0000
+        assert "ErrorComment" not in rsp
+        assoc.release()
+        assert assoc.is_released
+
+        assert attrs["uid"] == NewStorage
+
+        scp.shutdown()
diff --git a/pynetdicom/tests/test_service_substance_admin.py b/pynetdicom/tests/test_service_substance_admin.py
index 4a7af6e21..6b72a5553 100644
--- a/pynetdicom/tests/test_service_substance_admin.py
+++ b/pynetdicom/tests/test_service_substance_admin.py
@@ -8,7 +8,7 @@ import pytest
 from pydicom.dataset import Dataset
 from pydicom.uid import ExplicitVRLittleEndian
 
-from pynetdicom import AE, evt, debug_logger
+from pynetdicom import AE, evt, debug_logger, register_uid, sop_class
 from pynetdicom.dimse_primitives import C_FIND
 from pynetdicom.presentation import PresentationContext
 from pynetdicom.service_class import (
@@ -16,6 +16,7 @@ from pynetdicom.service_class import (
 )
 from pynetdicom.sop_class import (
     ProductCharacteristicsQuery,
+    _SUBSTANCE_ADMINISTRATION_CLASSES,
 )
 
 
@@ -36,6 +37,20 @@ def test_unknown_sop_class():
         service.SCP(C_FIND(), context)
 
 
+@pytest.fixture()
+def register_new_uid():
+    register_uid(
+        "1.2.3.4",
+        "NewFind",
+        SubstanceAdministrationQueryServiceClass,
+        "C-FIND",
+    )
+    yield
+    del _SUBSTANCE_ADMINISTRATION_CLASSES["NewFind"]
+    delattr(sop_class, "NewFind")
+    SubstanceAdministrationQueryServiceClass._SUPPORTED_UIDS["C-FIND"].remove("1.2.3.4")
+
+
 class TestSubstanceAdministrationQueryServiceClass:
     """Test the SubstanceAdministrationQueryServiceClass.
 
@@ -230,3 +245,35 @@ class TestSubstanceAdministrationQueryServiceClass:
         assoc.release()
         assert assoc.is_released
         scp.shutdown()
+
+    def test_register(self, register_new_uid):
+        """Test handler yielding a Dataset status"""
+        from pynetdicom.sop_class import NewFind
+
+        def handle(event):
+            status = Dataset()
+            status.Status = 0xFF00
+            yield status, self.query
+            yield 0x0000, None
+
+        handlers = [(evt.EVT_C_FIND, handle)]
+
+        self.ae = ae = AE()
+        ae.add_supported_context(NewFind)
+        ae.add_requested_context(NewFind, ExplicitVRLittleEndian)
+        scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
+
+        assoc = ae.associate("localhost", 11112)
+        assert assoc.is_established
+        result = assoc.send_c_find(self.query, NewFind)
+        status, identifier = next(result)
+        assert status.Status == 0xFF00
+        status, identifier = next(result)
+        assert status.Status == 0x0000
+        with pytest.raises(StopIteration):
+            next(result)
+
+        assoc.release()
+        assert assoc.is_released
+
+        scp.shutdown()
diff --git a/pynetdicom/tests/test_sop.py b/pynetdicom/tests/test_sop.py
index 88fafb7e0..213d718c2 100644
--- a/pynetdicom/tests/test_sop.py
+++ b/pynetdicom/tests/test_sop.py
@@ -5,6 +5,7 @@ import pytest
 from pynetdicom import __version__
 from pydicom._uid_dict import UID_dictionary
 
+from pynetdicom import sop_class
 from pynetdicom.sop_class import (
     uid_to_sop_class,
     uid_to_service_class,
@@ -64,6 +65,7 @@ from pynetdicom.sop_class import (
     SubstanceAdministrationLoggingInstance,
     UPSFilteredGlobalSubscriptionInstance,
     UPSGlobalSubscriptionInstance,
+    register_uid,
 )
 from pynetdicom.service_class import (
     ServiceClass,
@@ -166,11 +168,11 @@ def test_all_sop_instances():
 class TestUIDtoSOPlass:
     """Tests for uid_to_sop_class"""
 
-    def test_missing_sop(self):
+    def test_missing_sop_class(self):
         """Test SOP Class if UID not found."""
-        sop_class = uid_to_sop_class("1.2.3.4")
-        assert sop_class == "1.2.3.4"
-        assert sop_class.service_class == ServiceClass
+        sop = uid_to_sop_class("1.2.3.4")
+        assert sop == "1.2.3.4"
+        assert sop.service_class == ServiceClass
 
     def test_verification_uid(self):
         """Test normal function"""
@@ -179,8 +181,8 @@ class TestUIDtoSOPlass:
     def test_existing(self):
         """Test that the existing class is returned."""
         original = Verification
-        sop_class = uid_to_sop_class("1.2.840.10008.1.1")
-        assert id(sop_class) == id(original)
+        sop = uid_to_sop_class("1.2.840.10008.1.1")
+        assert id(sop) == id(original)
 
 
 class TestUIDToServiceClass:
@@ -471,13 +473,158 @@ class TestSOPClass:
 
     def test_uid_creation(self):
         """Test creating a new UIDSOPClass."""
-        sop_class = SOPClass("1.2.3")
-        sop_class._service_class = ServiceClass
+        sop = SOPClass("1.2.3")
+        sop._service_class = ServiceClass
 
-        assert sop_class == "1.2.3"
-        assert sop_class.service_class == ServiceClass
+        assert sop == "1.2.3"
+        assert sop.service_class == ServiceClass
 
-        sop_class_b = SOPClass(sop_class)
-        assert sop_class == sop_class_b
-        assert sop_class_b == "1.2.3"
-        assert sop_class_b.service_class == ServiceClass
+        sop_b = SOPClass(sop)
+        assert sop == sop_b
+        assert sop_b == "1.2.3"
+        assert sop_b.service_class == ServiceClass
+
+
+class TestRegisterUID:
+    def test_register_storage(self):
+        """Test registering to the storage service."""
+        register_uid(
+            "1.2.3.4",
+            "FooStorage",
+            StorageServiceClass,
+        )
+
+        sop = sop_class.FooStorage
+        assert sop == "1.2.3.4"
+        assert sop.service_class == StorageServiceClass
+
+        del _STORAGE_CLASSES["FooStorage"]
+        delattr(sop_class, "FooStorage")
+
+    def test_register_qr_find(self):
+        """Test registering to the QR service - FIND."""
+        register_uid(
+            "1.2.3.4",
+            "FooFind",
+            QueryRetrieveServiceClass,
+            dimse_msg_type="C-FIND",
+        )
+
+        sop = sop_class.FooFind
+        assert sop == "1.2.3.4"
+        assert sop.service_class == QueryRetrieveServiceClass
+        assert sop in QueryRetrieveServiceClass._SUPPORTED_UIDS["C-FIND"]
+
+        del _QR_CLASSES["FooFind"]
+        QueryRetrieveServiceClass._SUPPORTED_UIDS["C-FIND"].remove(sop)
+        delattr(sop_class, "FooFind")
+
+    def test_register_qr_get(self):
+        """Test registering to the QR service - GET."""
+        register_uid(
+            "1.2.3.4",
+            "FooGet",
+            QueryRetrieveServiceClass,
+            dimse_msg_type="C-GET",
+        )
+
+        sop = sop_class.FooGet
+        assert sop == "1.2.3.4"
+        assert sop.service_class == QueryRetrieveServiceClass
+        assert sop in QueryRetrieveServiceClass._SUPPORTED_UIDS["C-GET"]
+
+        del _QR_CLASSES["FooGet"]
+        QueryRetrieveServiceClass._SUPPORTED_UIDS["C-GET"].remove(sop)
+        delattr(sop_class, "FooGet")
+
+    def test_register_qr_move(self):
+        """Test registering to the QR service - MOVE."""
+        register_uid(
+            "1.2.3.4",
+            "FooMove",
+            QueryRetrieveServiceClass,
+            dimse_msg_type="C-MOVE",
+        )
+
+        sop = sop_class.FooMove
+        assert sop == "1.2.3.4"
+        assert sop.service_class == QueryRetrieveServiceClass
+        assert sop in QueryRetrieveServiceClass._SUPPORTED_UIDS["C-MOVE"]
+
+        del _QR_CLASSES["FooMove"]
+        QueryRetrieveServiceClass._SUPPORTED_UIDS["C-MOVE"].remove(sop)
+        delattr(sop_class, "FooMove")
+
+    def test_register_bwm_find(self):
+        """Test registering to the BWM service."""
+        register_uid(
+            "1.2.3.4",
+            "FooFind",
+            BasicWorklistManagementServiceClass,
+            dimse_msg_type="C-FIND",
+        )
+
+        sop = sop_class.FooFind
+        assert sop == "1.2.3.4"
+        assert sop.service_class == BasicWorklistManagementServiceClass
+        assert sop in BasicWorklistManagementServiceClass._SUPPORTED_UIDS["C-FIND"]
+
+        del _BASIC_WORKLIST_CLASSES["FooFind"]
+        BasicWorklistManagementServiceClass._SUPPORTED_UIDS["C-FIND"].remove(sop)
+        delattr(sop_class, "FooFind")
+
+    def test_register_substance_admin_find(self):
+        """Test registering to the Substance Admin QR service."""
+        register_uid(
+            "1.2.3.4",
+            "FooFind",
+            SubstanceAdministrationQueryServiceClass,
+            dimse_msg_type="C-FIND",
+        )
+
+        sop = sop_class.FooFind
+        assert sop == "1.2.3.4"
+        assert sop.service_class == SubstanceAdministrationQueryServiceClass
+        assert sop in SubstanceAdministrationQueryServiceClass._SUPPORTED_UIDS["C-FIND"]
+
+        del _SUBSTANCE_ADMINISTRATION_CLASSES["FooFind"]
+        SubstanceAdministrationQueryServiceClass._SUPPORTED_UIDS["C-FIND"].remove(sop)
+        delattr(sop_class, "FooFind")
+
+    def test_invalid_keyword_raises(self):
+        """Test invalid keyword raises exceptions."""
+        msg = (
+            "The keyword '2coo' is not a valid Python identifier or is "
+            "a Python keyword"
+        )
+        with pytest.raises(ValueError, match=msg):
+            register_uid("", "2coo", ServiceClass)
+
+        msg = (
+            "The keyword 'def' is not a valid Python identifier or is "
+            "a Python keyword"
+        )
+        with pytest.raises(ValueError, match=msg):
+            register_uid("", "def", ServiceClass)
+
+    def test_invalid_service_class_raises(self):
+        """Test that an invalid service_class raises exceptions."""
+        msg = "'service_class' must be a class object not a class instance"
+        with pytest.raises(TypeError, match=msg):
+            register_uid("", "Foo", "")
+
+        msg = (
+            "'service_class' must be a ServiceClass subclass object "
+            "such as 'StorageServiceClass'"
+        )
+        with pytest.raises(TypeError, match=msg):
+            register_uid("", "Foo", str)
+
+    def test_invalid_dimse_msg_type_raises(self):
+        """Test that an invalid dimse_msg_type raises exceptions."""
+        msg = (
+            "'dimse_msg_type' must be 'C-FIND', 'C-GET' or 'C-MOVE' "
+            "when registering a UID with QueryRetrieveServiceClass"
+        )
+        with pytest.raises(ValueError, match=msg):
+            register_uid("", "Foo", QueryRetrieveServiceClass, "Foo")