new operation get_library_class_domains

cdisc_rules_engine/models/operation_params.py

@@ -38,6 +38,7 @@ class OperationParams:
     ct_package_types: List[str] = None
     ct_version: str = None
     ct_package_type: str = None
+    domain_class: str = None
     term_code: str = None
     term_value: str = None
     term_pref_term: str = None

cdisc_rules_engine/operations/get_library_class_domains.py

@@ -0,0 +1,29 @@
+from cdisc_rules_engine.operations.base_operation import BaseOperation
+
+
+class GetLibraryClassDomains(BaseOperation):
+    """
+    A class for fetching domains for a given class from the CDISC Library.
+    Retrieves the list of domains based on standard, version, and optional class filter.
+    Example use case: FB1101 - "All Trial Design datasets as listed in the corresponding IG should be submitted"
+    """
+
+    def _execute_operation(self):
+        domain_class = getattr(self.params, "domain_class", None)
+        standard_details = self.library_metadata.standard_metadata
+        domains = self._extract_domains(standard_details, domain_class)
+        return domains
+
+    def _extract_domains(self, standard_details: dict, domain_class: str = None) -> set:
+        domains = set()
+        classes = standard_details.get("classes", [])
+
+        for cls in classes:
+            if domain_class and cls.get("name") != domain_class:
+                continue
+            datasets = cls.get("datasets", [])
+            for dataset in datasets:
+                domain_name = dataset.get("name")
+                domains.add(domain_name)
+
+        return domains

cdisc_rules_engine/utilities/rule_processor.py

@@ -352,54 +352,55 @@ def perform_rule_operations(
 
             # get necessary operation
             operation_params = OperationParams(
+                attribute_name=operation.get("attribute_name", ""),
+                case_sensitive=operation.get("case_sensitive", True),
+                codelist=operation.get("codelist"),
+                codelist_code=operation.get("codelist_code"),
+                codelists=operation.get("codelists"),
                 core_id=rule.get("core_id"),
-                operation_id=operation.get("id"),
-                operation_name=operation.get("operator"),
-                dataframe=dataset_copy,
-                target=target,
-                original_target=original_target,
-                domain=domain,
-                dataset_path=dataset_path,
-                directory_path=get_directory_path(dataset_path),
-                datasets=datasets,
-                grouping=operation.get("group", []),
-                standard=standard,
-                standard_version=standard_version,
-                standard_substandard=standard_substandard,
-                external_dictionaries=external_dictionaries,
-                ct_version=operation.get("version"),
+                ct_attribute=operation.get("ct_attribute"),
                 ct_package_type=RuleProcessor._ct_package_type_api_name(
                     operation.get("ct_package_type")
                 ),
-                ct_attribute=operation.get("ct_attribute"),
                 ct_package_types=[
                     RuleProcessor._ct_package_type_api_name(ct_package_type)
                     for ct_package_type in operation.get("ct_package_types", [])
                 ],
-                attribute_name=operation.get("attribute_name", ""),
-                key_name=operation.get("key_name", ""),
-                key_value=operation.get("key_value", ""),
-                case_sensitive=operation.get("case_sensitive", True),
-                external_dictionary_type=operation.get("external_dictionary_type"),
+                ct_version=operation.get("version"),
+                dataframe=dataset_copy,
+                dataset_path=dataset_path,
+                datasets=datasets,
+                delimiter=operation.get("delimiter"),
+                dictionary_term_type=operation.get("dictionary_term_type"),
+                directory_path=get_directory_path(dataset_path),
+                domain=domain,
+                domain_class=operation.get("domain_class"),
+                external_dictionaries=external_dictionaries,
                 external_dictionary_term_variable=operation.get(
                     "external_dictionary_term_variable"
                 ),
-                dictionary_term_type=operation.get("dictionary_term_type"),
+                external_dictionary_type=operation.get("external_dictionary_type"),
                 filter=operation.get("filter", None),
+                grouping=operation.get("group", []),
                 grouping_aliases=operation.get("group_aliases"),
+                key_name=operation.get("key_name", ""),
+                key_value=operation.get("key_value", ""),
                 level=operation.get("level"),
-                returntype=operation.get("returntype"),
-                codelists=operation.get("codelists"),
-                codelist=operation.get("codelist"),
-                codelist_code=operation.get("codelist_code"),
                 map=operation.get("map"),
+                namespace=operation.get("namespace"),
+                operation_id=operation.get("id"),
+                operation_name=operation.get("operator"),
+                original_target=original_target,
+                regex=operation.get("regex"),
+                returntype=operation.get("returntype"),
+                standard=standard,
+                standard_substandard=standard_substandard,
+                standard_version=standard_version,
+                target=target,
                 term_code=operation.get("term_code"),
-                term_value=operation.get("term_value"),
                 term_pref_term=operation.get("term_pref_term"),
-                namespace=operation.get("namespace"),
+                term_value=operation.get("term_value"),
                 value_is_reference=operation.get("value_is_reference", False),
-                delimiter=operation.get("delimiter"),
-                regex=operation.get("regex"),
             )
             try:
                 # execute operation

resources/schema/rule/Operations.json

@@ -108,6 +108,13 @@
       "required": ["id", "operator"],
       "type": "object"
     },
+    {
+      "properties": {
+        "operator": { "const": "get_library_class_domains" }
+      },
+      "required": ["id", "operator"],
+      "type": "object"
+    },
     {
       "properties": {
         "operator": {

resources/schema/rule/Operations.md

@@ -846,6 +846,16 @@ Output
 }
 ```
 
+### get_library_class_domains
+
+Returns the list of domain for a given class from the CDISC Library Implementation Guide. This operation retrieves all domains that belong to a specified class (e.g., "TRIAL DESIGN", "FINDINGS", "EVENTS") based on the current standard and version. The operation uses the standard and version from the validation context as well as the optional `domain_class` parameter which is the name of the class to filter by (e.g., "TRIAL DESIGN", "FINDINGS", "EVENTS", "INTERVENTIONS). NOTE: Class names are case-sensitive and should match the Library metadata format. If no `domain_class` parameter is provided, the operation returns all domains across all classes in the Implementation Guide:
+
+```yaml
+- operator: get_library_class_domains
+  id: $trial_design_domains
+  domain_class: "TRIAL DESIGN"
+```
+
 ## Define.XML Metadata Operations
 
 Operations for working with Define.XML metadata and variable references.

tests/unit/test_operations/test_get_library_class_domains.py

@@ -0,0 +1,157 @@
+from cdisc_rules_engine.config.config import ConfigService
+from cdisc_rules_engine.models.library_metadata_container import (
+    LibraryMetadataContainer,
+)
+from cdisc_rules_engine.models.operation_params import OperationParams
+from cdisc_rules_engine.operations.get_library_class_domains import (
+    GetLibraryClassDomains,
+)
+from cdisc_rules_engine.services.cache import InMemoryCacheService
+from cdisc_rules_engine.services.data_services import LocalDataService
+
+
+def test_get_library_class_domains(operation_params: OperationParams):
+    """
+    Test that get_library_class_domains returns the correct domains for a given class.
+    """
+    standard_metadata = {
+        "classes": [
+            {
+                "name": "TRIAL DESIGN",
+                "datasets": [
+                    {"name": "TA"},
+                    {"name": "TE"},
+                    {"name": "TI"},
+                    {"name": "TS"},
+                    {"name": "TV"},
+                ],
+            },
+            {
+                "name": "EVENTS",
+                "datasets": [
+                    {"name": "AE"},
+                    {"name": "CE"},
+                    {"name": "DS"},
+                ],
+            },
+            {
+                "name": "FINDINGS",
+                "datasets": [
+                    {"name": "LB"},
+                    {"name": "VS"},
+                    {"name": "EG"},
+                ],
+            },
+        ]
+    }
+
+    operation_params.standard = "sdtmig"
+    operation_params.standard_version = "3-4"
+    operation_params.domain_class = "TRIAL DESIGN"
+
+    cache = InMemoryCacheService.get_instance()
+    library_metadata = LibraryMetadataContainer(standard_metadata=standard_metadata)
+    data_service = LocalDataService.get_instance(
+        cache_service=cache, config=ConfigService()
+    )
+
+    operation = GetLibraryClassDomains(
+        operation_params,
+        operation_params.dataframe,
+        cache,
+        data_service,
+        library_metadata,
+    )
+
+    domains = operation._execute_operation()
+
+    expected_domains = {"TA", "TE", "TI", "TS", "TV"}
+    assert domains == expected_domains, (
+        f"Domain mismatch:\n" f"  Expected: {expected_domains}\n" f"  Got: {domains}"
+    )
+
+
+def test_get_library_class_domains_no_filter(operation_params: OperationParams):
+    """
+    Test that get_library_class_domains returns all domains when no class filter is provided.
+    """
+    standard_metadata = {
+        "classes": [
+            {
+                "name": "TRIAL DESIGN",
+                "datasets": [
+                    {"name": "TA"},
+                    {"name": "TE"},
+                ],
+            },
+            {
+                "name": "EVENTS",
+                "datasets": [
+                    {"name": "AE"},
+                    {"name": "DS"},
+                ],
+            },
+        ]
+    }
+
+    operation_params.standard = "sdtmig"
+    operation_params.standard_version = "3-4"
+
+    cache = InMemoryCacheService.get_instance()
+    library_metadata = LibraryMetadataContainer(standard_metadata=standard_metadata)
+    data_service = LocalDataService.get_instance(
+        cache_service=cache, config=ConfigService()
+    )
+
+    operation = GetLibraryClassDomains(
+        operation_params,
+        operation_params.dataframe,
+        cache,
+        data_service,
+        library_metadata,
+    )
+
+    domains = operation._execute_operation()
+
+    expected_domains = {"TA", "TE", "AE", "DS"}
+    assert domains == expected_domains, (
+        f"Domain mismatch:\n" f"  Expected: {expected_domains}\n" f"  Got: {domains}"
+    )
+
+
+def test_get_library_class_domains_empty_class(operation_params: OperationParams):
+    """
+    Test that get_library_class_domains returns empty set for non-existent class.
+    """
+    standard_metadata = {
+        "classes": [
+            {
+                "name": "TRIAL DESIGN",
+                "datasets": [
+                    {"name": "TA"},
+                ],
+            },
+        ]
+    }
+
+    operation_params.standard = "sdtmig"
+    operation_params.standard_version = "3-4"
+    operation_params.domain_class = "Nonexistent Class"
+
+    cache = InMemoryCacheService.get_instance()
+    library_metadata = LibraryMetadataContainer(standard_metadata=standard_metadata)
+    data_service = LocalDataService.get_instance(
+        cache_service=cache, config=ConfigService()
+    )
+
+    operation = GetLibraryClassDomains(
+        operation_params,
+        operation_params.dataframe,
+        cache,
+        data_service,
+        library_metadata,
+    )
+
+    domains = operation._execute_operation()
+
+    assert domains == set(), f"Expected empty set, got: {domains}"