refactor(Clients): Replace config_get with ConfigOption calls

Assisted by Claude Sonnet 4.6 - used to find and replace without logic changes.

Issue: documentation#325

Author: voetberg

bin/rucio-auditor

@@ -29,6 +29,7 @@ import rucio.common.config as config
 import rucio.common.dumper as dumper
 import rucio.daemons.auditor
 from rucio.client.rseclient import RSEClient
+from rucio.common.config_settings import Config
 from rucio.common.exception import RSENotFound
 
 RETRY_AFTER = 60 * 60 * 24 * 14  # Two weeks
@@ -68,7 +69,7 @@ def main(args):
     terminate = Event()
     logpipes = []
 
-    loglevel = logging.getLevelName(config.config_get('common', 'loglevel'))
+    loglevel = logging.getLevelName(Config.common.loglevel())
 
     mainlogr, mainlogw = Pipe(duplex=False)
     logpipes.append(mainlogr)
@@ -77,8 +78,8 @@ def main(args):
     if not config.config_has_section('auditor'):
         raise NoSectionError("Auditor section required in config to run the auditor daemon.")
 
-    cache_dir = config.config_get('auditor', 'cache')
-    results_dir = config.config_get('auditor', 'results')
+    cache_dir = Config.auditor.cache()
+    results_dir = Config.auditor.results()
 
     logfilename = os.path.join(config.config_get('common', 'logdir'), 'auditor.log')
     logger.info('Starting auditor')

lib/rucio/cli/bin_legacy/rucio.py

@@ -39,7 +39,8 @@
 from rucio.cli.utils import exception_handler, get_client, get_scope, setup_gfal2_logger, signal_handler
 from rucio.client.richclient import MAX_TRACEBACK_WIDTH, MIN_CONSOLE_WIDTH, CLITheme, generate_table, get_cli_config, get_pager, print_output, setup_rich_logger
 from rucio.common.client import detect_client_location
-from rucio.common.config import config_get, config_get_float
+from rucio.common.config import config_get_float
+from rucio.common.config_settings import Config
 from rucio.common.constants import ReplicaState
 from rucio.common.exception import (
     DuplicateRule,
@@ -991,7 +992,7 @@ def download(args, client, logger, console, spinner):
         item_defaults['nrandom'] = args.nrandom
         item_defaults['transfer_speed_timeout'] = args.transfer_speed_timeout \
             if args.transfer_speed_timeout is not None \
-            else config_get_float('download', 'transfer_speed_timeout', False, 500)
+            else Config.download.transfer_speed_timeout(raise_exception=False)
         items = []
         if args.dids:
             for did in args.dids:
@@ -1117,7 +1118,7 @@ def get_metadata(args, client, logger, console, spinner):
     if args.plugin:
         plugin = args.plugin
     else:
-        plugin = config_get('client', 'metadata_default_plugin', default='DID_COLUMN')
+        plugin = Config.client.metadata_default_plugin()
 
     if cli_config == 'rich':
         spinner.update(status='Fetching metadata')

lib/rucio/cli/did.py

@@ -20,7 +20,7 @@
 
 from rucio.cli.utils import get_scope
 from rucio.client.richclient import CLITheme, generate_table, print_output
-from rucio.common.config import config_get
+from rucio.common.config_settings import Config
 from rucio.common.exception import InputValidationError, InvalidObject, RucioException, ScopeNotFound
 from rucio.common.utils import chunks, parse_did_filter_from_string_fe
 
@@ -389,7 +389,7 @@ def metadata_remove(ctx: click.Context, did: str, key: str) -> None:
 def metadata_list_(ctx: click.Context, dids: tuple[str, ...], plugin: Optional[str]) -> None:
     """List metadata for a list of DIDs"""
     if plugin is None:
-        plugin = config_get('client', 'metadata_default_plugin', default='DID_COLUMN')
+        plugin = Config.client.metadata_default_plugin()
 
     if ctx.obj.use_rich:
         ctx.obj.spinner.update(status='Fetching metadata')

lib/rucio/cli/utils.py

@@ -27,7 +27,7 @@
 import click
 
 from rucio.client.client import Client
-from rucio.common.config import config_get
+from rucio.common.config_settings import Config
 from rucio.common.exception import (
     AccessDenied,
     CannotAuthenticate,
@@ -124,7 +124,7 @@ def new_funct(*args, **kwargs):
             if "x-rucio-auth-token" in str(error):
                 used_account = None
                 try:  # get the configured account from the configuration file
-                    used_account = "%s (from rucio.cfg)" % config_get("client", "account")
+                    used_account = "%s (from rucio.cfg)" % Config.client.account()
                 except Exception:
                     pass
                 try:  # are we overriden by the environment?
@@ -135,7 +135,7 @@ def new_funct(*args, **kwargs):
 
             else:
                 logger.debug(traceback.format_exc())
-                contact = config_get("policy", "support", raise_exception=False)
+                contact = Config.policy.support(raise_exception=False)
                 support = ("Please follow up with all relevant information at: " + contact) if contact else ""
                 logger.error("\nThe object is missing this property: %s\n" 'This should never happen. Please rerun the last command with the "-v" option to gather more information.\n' "%s" % (str(error), support))
             return FAILURE
@@ -154,9 +154,9 @@ def new_funct(*args, **kwargs):
                 return SUCCESS
             logger.debug(traceback.format_exc())
             logger.error(error)
-            contact = config_get("policy", "support", raise_exception=False)
+            contact = Config.policy.support(raise_exception=False)
             support = ("If it's a problem concerning your experiment or if you're unsure what to do, please follow up at: %s\n" % contact) if contact else ""
-            contact = config_get("policy", "support_rucio", default="https://github.com/rucio/rucio/issues")
+            contact = Config.policy.support_rucio()
             support += "If you're sure there is a problem with Rucio itself, please follow up at: " + contact
             logger.error("\nRucio exited with an unexpected/unknown error.\n" 'Please rerun the last command with the "-v" option to gather more information.\n' "%s" % support)
             return FAILURE
@@ -179,7 +179,7 @@ def get_client(args, logger):
             auth_type = os.environ["RUCIO_AUTH_TYPE"].lower()
         else:
             try:
-                auth_type = config_get("client", "auth_type").lower()
+                auth_type = Config.client.auth_type().lower()
             except (NoOptionError, NoSectionError):
                 logger.error("Cannot get AUTH_TYPE")
                 sys.exit(1)

lib/rucio/client/baseclient.py

@@ -39,6 +39,7 @@
 from rucio import version
 from rucio.common import exception
 from rucio.common.config import config_get, config_get_bool, config_get_int, config_has_section
+from rucio.common.config_settings import Config
 from rucio.common.constants import DEFAULT_VO, HTTPMethod
 from rucio.common.exception import CannotAuthenticate, ClientProtocolNotFound, ClientProtocolNotSupported, ConfigNotFound, MissingClientParameter, MissingModuleException, NoAuthInformation, ServerConnectionException
 from rucio.common.extra import import_extras
@@ -141,15 +142,15 @@ def __init__(self,
             if rucio_host is not None:
                 self.host = rucio_host
             else:
-                self.host = config_get('client', 'rucio_host')
+                self.host = Config.client.rucio_host()
         except (NoOptionError, NoSectionError) as error:
             raise MissingClientParameter('Section client and Option \'%s\' cannot be found in config file' % error.args[0])
 
         try:
             if auth_host is not None:
                 self.auth_host = auth_host
             else:
-                self.auth_host = config_get('client', 'auth_host')
+                self.auth_host = Config.client.auth_host()
         except (NoOptionError, NoSectionError) as error:
             raise MissingClientParameter('Section client and Option \'%s\' cannot be found in config file' % error.args[0])
 
@@ -197,7 +198,7 @@ def __init__(self,
             if self.ca_cert is None:
                 self.logger.debug('HTTPS is required, but no ca_cert was passed and X509_CERT_DIR is not defined. Trying to get it from the config file.')
                 try:
-                    self.ca_cert = _expand_path(config_get('client', 'ca_cert'))
+                    self.ca_cert = _expand_path(Config.client.ca_cert())
                 except (NoOptionError, NoSectionError):
                     self.logger.debug('No ca_cert found in configuration. Falling back to Mozilla default CA bundle (certifi).')
                     self.ca_cert = True
@@ -211,7 +212,7 @@ def __init__(self,
                 self.account = environ['RUCIO_ACCOUNT']
             except KeyError:
                 try:
-                    self.account = config_get('client', 'account')
+                    self.account = Config.client.account()
                 except (NoOptionError, NoSectionError):
                     pass
 
@@ -224,7 +225,7 @@ def __init__(self,
             except KeyError:
                 self.logger.debug('No VO found. Trying to get it from the config file.')
                 try:
-                    self.vo = config_get('client', 'vo')
+                    self.vo = Config.client.vo()
                 except (NoOptionError, NoSectionError):
                     self.logger.debug('No VO found. Using default VO.')
                     self.vo = DEFAULT_VO
@@ -244,7 +245,7 @@ def __init__(self,
 
     def _get_auth_tokens(self) -> tuple[Optional[str], str, str, str]:
         # if token file path is defined in the rucio.cfg file, use that file. Currently this prevents authenticating as another user or VO.
-        auth_token_file_path = config_get('client', 'auth_token_file_path', False, None)
+        auth_token_file_path = Config.client.auth_token_file_path(raise_exception=False)
         token_filename_suffix = "for_default_account" if self.account is None else "for_account_" + self.account
 
         if auth_token_file_path:
@@ -270,7 +271,7 @@ def _get_auth_type(self, auth_type: Optional[str]) -> str:
                 auth_type = environ['RUCIO_AUTH_TYPE']
             else:
                 try:
-                    auth_type = config_get('client', 'auth_type')
+                    auth_type = Config.client.auth_type()
                 except (NoOptionError, NoSectionError) as error:
                     raise MissingClientParameter('Option \'%s\' cannot be found in config file' % error.args[0])
         return auth_type
@@ -288,7 +289,7 @@ def _get_creds(self, creds: Optional[dict[str, Any]]) -> dict[str, Any]:
                 if 'oidc_issuer' not in creds or creds['oidc_issuer'] is None:
                     creds['oidc_issuer'] = config_get('client', 'oidc_issuer', False, None)
                 if 'oidc_audience' not in creds or creds['oidc_audience'] is None:
-                    creds['oidc_audience'] = config_get('client', 'oidc_audience', False, None)
+                    creds['oidc_audience'] = Config.client.oidc_audience(raise_exception=False)
                 if 'oidc_auto' not in creds or creds['oidc_auto'] is False:
                     creds['oidc_auto'] = config_get_bool('client', 'oidc_auto', False, False)
                 if creds['oidc_auto']:
@@ -297,22 +298,22 @@ def _get_creds(self, creds: Optional[dict[str, Any]]) -> dict[str, Any]:
                     if 'oidc_password' not in creds or creds['oidc_password'] is None:
                         creds['oidc_password'] = config_get('client', 'oidc_password', False, None)
                 if 'oidc_scope' not in creds or creds['oidc_scope'] == 'openid profile':
-                    creds['oidc_scope'] = config_get('client', 'oidc_scope', False, 'openid profile')
+                    creds['oidc_scope'] = Config.client.oidc_scope()
                 if 'oidc_polling' not in creds or creds['oidc_polling'] is False:
                     creds['oidc_polling'] = config_get_bool('client', 'oidc_polling', False, False)
 
             elif self.auth_type in ['userpass', 'saml']:
                 if 'username' not in creds or creds['username'] is None:
-                    creds['username'] = config_get('client', 'username')
+                    creds['username'] = Config.client.username()
                 if 'password' not in creds or creds['password'] is None:
-                    creds['password'] = config_get('client', 'password')
+                    creds['password'] = Config.client.password()
 
             elif self.auth_type == 'x509':
                 if 'client_cert' not in creds or creds['client_cert'] is None:
                     if "RUCIO_CLIENT_CERT" in environ:
                         creds['client_cert'] = environ["RUCIO_CLIENT_CERT"]
                     else:
-                        creds['client_cert'] = config_get('client', 'client_cert')
+                        creds['client_cert'] = Config.client.client_cert()
 
                 creds['client_cert'] = _expand_path(creds['client_cert'])
 
@@ -323,7 +324,7 @@ def _get_creds(self, creds: Optional[dict[str, Any]]) -> dict[str, Any]:
                     if "RUCIO_CLIENT_KEY" in environ:
                         creds['client_key'] = environ["RUCIO_CLIENT_KEY"]
                     else:
-                        creds['client_key'] = config_get('client', 'client_key')
+                        creds['client_key'] = Config.client.client_key()
 
                 creds['client_key'] = _expand_path(creds['client_key'])
                 if not os.path.exists(creds['client_key']):
@@ -346,8 +347,8 @@ def _get_creds(self, creds: Optional[dict[str, Any]]) -> dict[str, Any]:
                 if 'client_proxy' not in creds or creds['client_proxy'] is None:
                     if 'RUCIO_CLIENT_PROXY' in environ:
                         creds['client_proxy'] = environ['RUCIO_CLIENT_PROXY']
-                    elif config_has_section('client') and config_get('client', 'client_x509_proxy', default='') != '':
-                        creds['client_proxy'] = config_get('client', 'client_x509_proxy')
+                    elif config_has_section('client') and Config.client.client_x509_proxy(default='') != '':
+                        creds['client_proxy'] = Config.client.client_x509_proxy()
                     elif 'X509_USER_PROXY' in environ:
                         creds['client_proxy'] = environ['X509_USER_PROXY']
                     elif os.path.isfile(gsi_proxy_path):
@@ -363,7 +364,7 @@ def _get_creds(self, creds: Optional[dict[str, Any]]) -> dict[str, Any]:
 
             elif self.auth_type == 'ssh':
                 if 'ssh_private_key' not in creds or creds['ssh_private_key'] is None:
-                    creds['ssh_private_key'] = config_get('client', 'ssh_private_key')
+                    creds['ssh_private_key'] = Config.client.ssh_private_key()
 
                 creds['ssh_private_key'] = _expand_path(creds['ssh_private_key'])
                 if not os.path.isfile(creds["ssh_private_key"]):

lib/rucio/client/downloadclient.py

@@ -32,6 +32,7 @@
 from rucio.common.checksum import CHECKSUM_ALGO_DICT, GLOBALLY_SUPPORTED_CHECKSUMS, PREFERRED_CHECKSUM, adler32
 from rucio.common.client import detect_client_location
 from rucio.common.config import config_get
+from rucio.common.config_settings import Config
 from rucio.common.constants import DEFAULT_VO
 from rucio.common.didtype import DID
 from rucio.common.exception import InputValidationError, NoFilesDownloaded, NotAllFilesDownloaded, RucioException
@@ -229,7 +230,7 @@ def __init__(
         # tar -C <dest_dir_path> -xf <archive_file_path>  <did_name>
         extract_args = '-C %(dest_dir_path)s -xf %(archive_file_path)s %(file_to_extract)s'
         self.extraction_tools.append(BaseExtractionTool('tar', '--version', extract_args, logger=self.logger))
-        self.extract_scope_convention = config_get('common', 'extract_scope', False, None)
+        self.extract_scope_convention = Config.common.extract_scope(raise_exception=False)
 
     def download_pfns(
         self,

lib/rucio/client/uploadclient.py

@@ -29,7 +29,8 @@
 from rucio.common.bittorrent import bittorrent_v2_merkle_sha256
 from rucio.common.checksum import GLOBALLY_SUPPORTED_CHECKSUMS, adler32, md5
 from rucio.common.client import detect_client_location
-from rucio.common.config import config_get, config_get_bool, config_get_int
+from rucio.common.config import config_get, config_get_bool
+from rucio.common.config_settings import Config
 from rucio.common.constants import DEFAULT_VO, RseAttr
 from rucio.common.exception import (
     DataIdentifierAlreadyExists,
@@ -1230,7 +1231,7 @@ def _retry_protocol_stat(
         Exception
             If the requested service is not available or permissions are not granted.
         """
-        retries = config_get_int('client', 'protocol_stat_retries', raise_exception=False, default=6)
+        retries = Config.client.protocol_stat_retries(raise_exception=False)
         for attempt in range(retries):
             try:
                 self.logger(logging.DEBUG, 'stat: pfn=%s' % pfn)