FirecREST¶

import json
import firecrest as f7t

client_id = "<client_id>"
client_secret = "<client_secret>"
token_uri = "https://auth.cscs.ch/auth/realms/firecrest-clients/protocol/openid-connect/token"

# Setup the client for the specific account
# For instance, for the Alps HPC Platform system Daint:

client = f7t.v2.Firecrest(
    firecrest_url="https://api.cscs.ch/hpc/firecrest/v2",
    authorization=f7t.ClientCredentialsAuth(client_id, client_secret, token_uri)
)

# Status of the systems, filesystems and schedulers:
print(json.dumps(client.systems(), indent=2))

# Output: information about systems and health status of the infrastructure
# [
#   {
#     "name": "daint",
#     "ssh": {                           # --> SSH settings
#       "host": "daint.alps.cscs.ch",
#       "port": 22,
#       "maxClients": 100,
#       "timeout": {
#         "connection": 5,
#         "login": 5,
#         "commandExecution": 5,
#         "idleTimeout": 60,
#         "keepAlive": 5
#       }
#     },
#     "scheduler": {                     # --> Scheduler settings
#       "type": "slurm",
#       "version": "24.05.4",
#       "apiUrl": null,
#       "apiVersion": null,
#       "timeout": 10
#     },
#     "servicesHealth": [                # --> Health status of services
#       {
#         "serviceType": "scheduler",
#         "lastChecked": "2025-03-18T23:34:51.167545Z",
#         "latency": 0.4725925922393799,
#         "healthy": true,
#         "message": null,
#         "nodes": {
#           "available": 21,
#           "total": 858
#         }
#       },
#       {
#         "serviceType": "ssh",
#         "lastChecked": "2025-03-18T23:34:52.054056Z",
#         "latency": 1.358715295791626,
#         "healthy": true,
#         "message": null
#       },
#       {
#         "serviceType": "filesystem",
#         "lastChecked": "2025-03-18T23:34:51.969350Z",
#         "latency": 1.2738196849822998,
#         "healthy": true,
#         "message": null,
#         "path": "/capstor/scratch/cscs"
#       },
#     (...)
#     "fileSystems": [                   # --> Filesystem settings
#       {
#         "path": "/capstor/scratch/cscs",
#         "dataType": "scratch",
#         "defaultWorkDir": true
#       },
#       {
#         "path": "/users",
#         "dataType": "users",
#         "defaultWorkDir": false
#       },
#       {
#         "path": "/capstor/store/cscs",
#         "dataType": "store",
#         "defaultWorkDir": false
#       }
#     ]    
#   }
# ]

# List content of directories
print(json.dumps(client.list_files("daint", "/capstor/scratch/cscs/<username>"),
                                indent=2))

# [
#   {
#     "name": "directory",
#     "type": "d",
#     "linkTarget": null,
#     "user": "<username>",
#     "group": "<project>",
#     "permissions": "rwxr-x---+",
#     "lastModified": "2024-09-02T12:34:45",
#     "size": "4096"
#   },
#   {
#     "name": "file.txt",
#     "type": "-",
#     "linkTarget": null,
#     "user": "<username>",
#     "group": "<project>",
#     "permissions": "rw-r-----+",
#     "lastModified": "2024-09-02T08:26:04",
#     "size": "131225"
#   }
# ]

import json
import os
import sys

import httpx
from firecrest.v2 import Firecrest


DEFAULT_URL = "https://f7t-pat.api.svc.cscs.ch/hpcp"
API_KEY_HEADER = "X-API-Key"

API_KEY = os.environ.get("CSCS_API_KEY")
if not API_KEY:
    print("Set the CSCS_API_KEY environment variable")
    sys.exit(1)


class ApiKeyAuth:
    """Placeholder auth object that suppresses token-based authentication."""

    def __init__(self, api_key: str):
        self.api_key = api_key

    def get_access_token(self) -> str:
        return "unused-api-key-auth"


def _api_key_hook(api_key: str):
    """Return an httpx request hook that swaps bearer auth for the API key."""

    def hook(request: httpx.Request) -> None:
        request.headers.pop("Authorization", None)
        request.headers[API_KEY_HEADER] = api_key

    return hook


def create_client(
    api_key: str,
    firecrest_url: str = DEFAULT_URL,
) -> Firecrest:
    """Build a ``Firecrest`` client that authenticates with ``X-API-Key``."""
    client = Firecrest(
        firecrest_url=firecrest_url,
        authorization=ApiKeyAuth(api_key),
        verify=True,
    )

    hook = _api_key_hook(api_key)
    client._session.event_hooks["request"].append(hook)

    # close_session()/create_new_session() build a fresh httpx.Client, which
    # would come without our hook, so re-install it on every new session.
    original_create_new_session = client.create_new_session

    def create_new_session_with_hook() -> None:
        original_create_new_session()
        client._session.event_hooks["request"].append(hook)

    client.create_new_session = create_new_session_with_hook
    return client


SYSTEM = "daint"
client = create_client(api_key=API_KEY, firecrest_url=DEFAULT_URL)

#
# using the client
#

print(f"Server version: {client.server_version() or 'unknown'}")
systems = client.systems()
print(f"\nSystems ({len(systems)}):")
for system in systems:
    print(f"  - {system.get('name')}")

print("\nUser info:")
user = client.userinfo(system_name=SYSTEM)
print(json.dumps(user, indent=2))

print("\nHome directory:")
username = user["user"]["name"]
for entry in client.list_files(system_name=SYSTEM, path=f"/users/{username}"):
    print(f"  {entry.get('permissions', ''):>10}  {entry.get('name')}")

client.close_session()