feat: add RSS read adapter

src/DIRAC/RssReadAdapter/ResponseTranslation.py

@@ -0,0 +1,156 @@
+"""Response translation for the read adapter module.
+
+This module translates responses from diracx Pydantic models to legacy format.
+"""
+
+from __future__ import annotations
+
+from typing import Dict
+
+from diracx.core.models.rss import (
+    AllowedStatus,
+    BannedStatus,
+    ComputeElementStatus,
+    FTSStatus,
+    SiteStatus,
+    StorageElementStatus,
+)
+
+
+def translate_storage_element_status(
+    response: dict[str, StorageElementStatus],
+) -> list[tuple]:
+    """Translate storage element status from diracx format to legacy format.
+
+    Args:
+        response: Dictionary of storage element names to StorageElementStatus
+
+    Returns:
+        List of tuples in legacy format: (name, element_type, status_type, status, vo)
+
+    """
+    legacy_format = []
+    for name, status in response.items():
+        legacy_format.append(
+            (
+                name,
+                "StorageElement",
+                "ReadAccess",
+                _translate_resource_status(status.read),
+                None,
+            )
+        )
+        legacy_format.append(
+            (
+                name,
+                "StorageElement",
+                "WriteAccess",
+                _translate_resource_status(status.write),
+                None,
+            )
+        )
+        legacy_format.append(
+            (
+                name,
+                "StorageElement",
+                "CheckAccess",
+                _translate_resource_status(status.check),
+                None,
+            )
+        )
+        legacy_format.append(
+            (
+                name,
+                "StorageElement",
+                "RemoveAccess",
+                _translate_resource_status(status.remove),
+                None,
+            )
+        )
+
+    return legacy_format
+
+
+def translate_computing_element_status(
+    response: dict[str, ComputeElementStatus],
+) -> list[tuple]:
+    """Translate computing element status from diracx format to legacy format.
+
+    Args:
+        response: Dictionary of computing element names to ComputeElementStatus
+
+    Returns:
+        List of tuples in legacy format: (name, element_type, status_type, status, vo)
+
+    """
+    legacy_format = []
+    for name, status in response.items():
+        legacy_format.append(
+            (
+                name,
+                "ComputeElement",
+                "all",
+                _translate_resource_status(status.all),
+                None,
+            )
+        )
+    return legacy_format
+
+
+def translate_fts_status(response: dict[str, FTSStatus]) -> list[tuple]:
+    """Translate FTS server status from diracx format to legacy format.
+
+    Args:
+        response: Dictionary of FTS server names to FTSStatus
+
+    Returns:
+        List of tuples in legacy format: (name, element_type, status_type, status, vo)
+
+    """
+    legacy_format = []
+    for name, status in response.items():
+        legacy_format.append(
+            (
+                name,
+                "FTS",
+                "all",
+                _translate_resource_status(status.all),
+                None,
+            )
+        )
+    return legacy_format
+
+
+def translate_site_status(response: dict[str, SiteStatus]) -> list[tuple]:
+    """Translate site status from diracx format to legacy format.
+
+    Args:
+        response: Dictionary of site names to SiteStatus
+
+    Returns:
+        List of tuples in legacy format: (site, status)
+
+    """
+    legacy_format = []
+    for name, status in response.items():
+        legacy_format.append((name, _translate_resource_status(status.all)))
+    return legacy_format
+
+
+def _translate_resource_status(status: AllowedStatus | BannedStatus) -> str:
+    """Translate a single resource status from diracx format to legacy format.
+
+    Args:
+        status: ResourceStatus (AllowedStatus or BannedStatus)
+
+    Returns:
+        Legacy status string - only "Active" or "Banned" as per DIRAC streamlining
+
+    """
+    if isinstance(status, AllowedStatus):
+        # DIRAC is being streamlined to only use Active/Banned states
+        # All allowed statuses (including Degraded) are mapped to Active
+        return "Active"
+    else:  # BannedStatus
+        # All banned statuses are mapped to Banned
+        return "Banned"

src/DIRAC/RssReadAdapter/Statuses.py

@@ -0,0 +1,60 @@
+"""RSS API calling and apply translation for the read adapter module."""
+
+from __future__ import annotations
+
+from DIRAC.Core.Security.DiracX import DiracXClient
+
+from .ResponseTranslation import (
+    translate_computing_element_status,
+    translate_fts_status,
+    translate_site_status,
+    translate_storage_element_status,
+)
+
+
+def get_storage_element_status() -> list[tuple]:
+    """Get storage element status from the RSS API and translate to legacy format.
+
+    Returns:
+        List of tuples in legacy format: (name, element_type, status_type, status, vo)
+
+    """
+    with DiracXClient() as client:
+        response = client.rss.get_storage_status()
+        return translate_storage_element_status(response)
+
+
+def get_computing_element_status() -> list[tuple]:
+    """Get computing element status from the RSS API and translate to legacy format.
+
+    Returns:
+        List of tuples in legacy format: (name, element_type, status_type, status, vo)
+
+    """
+    with DiracXClient() as client:
+        response = client.rss.get_compute_status()
+        return translate_computing_element_status(response)
+
+
+def get_fts_status() -> list[tuple]:
+    """Get merged FTS server status from all VOs.
+
+    Returns:
+        List of tuples in legacy format: (name, element_type, status_type, status, vo)
+
+    """
+    with DiracXClient() as client:
+        response = client.rss.get_fts_status()
+        return translate_fts_status(response)
+
+
+def get_site_status() -> list[tuple]:
+    """Get site status from the RSS API and translate to legacy format.
+
+    Returns:
+        List of tuples in legacy format: (site, status)
+
+    """
+    with DiracXClient() as client:
+        response = client.rss.get_site_status()
+        return translate_site_status(response)

src/DIRAC/RssReadAdapter/__init__.py

@@ -0,0 +1,32 @@
+"""Read adapter module for translating diracx RSS API responses to legacy format.
+
+This module provides functionality to:
+- Call diracx RSS API endpoints and translate the outputs
+- Translate responses from new diracx format to legacy format
+"""
+
+from __future__ import annotations
+
+from .ResponseTranslation import (
+    translate_computing_element_status,
+    translate_fts_status,
+    translate_site_status,
+    translate_storage_element_status,
+)
+from .Statuses import (
+    get_computing_element_status,
+    get_fts_status,
+    get_site_status,
+    get_storage_element_status,
+)
+
+__all__ = [
+    "get_computing_element_status",
+    "get_fts_status",
+    "get_site_status",
+    "get_storage_element_status",
+    "translate_computing_element_status",
+    "translate_fts_status",
+    "translate_site_status",
+    "translate_storage_element_status",
+]