Source code for drms.config

from urllib.parse import urljoin

__all__ = ["ServerConfig", "register_server"]

[docs]class ServerConfig: """ DRMS Server configuration. Additional keyword arguments can be used to add additional entries to config. In case a keyword argument already exists in the config dictionary, the config entry will be replaced by the kwargs value. Available config keys are: name cgi_baseurl cgi_show_series cgi_jsoc_info cgi_jsoc_fetch cgi_check_address cgi_show_series_wrapper show_series_wrapper_dbhost url_show_series url_jsoc_info url_jsoc_fetch url_check_address url_show_series_wrapper encoding http_download_baseurl ftp_download_baseurl Parameters ---------- name : str Server configuration name. config : dict Dictionary containing configuration entries (see below for a list of available entries). """ _valid_keys = [ "name", "cgi_baseurl", "cgi_show_series", "cgi_jsoc_info", "cgi_jsoc_fetch", "cgi_check_address", "cgi_show_series_wrapper", "show_series_wrapper_dbhost", "url_show_series", "url_jsoc_info", "url_jsoc_fetch", "url_check_address", "url_show_series_wrapper", "encoding", "http_download_baseurl", "ftp_download_baseurl", ] def __init__(self, config=None, **kwargs): self._d = d = config.copy() if config is not None else {} d.update(kwargs) for k in d: if k not in self._valid_keys: raise ValueError(f"Invalid server config key: {k}") if "name" not in d: raise ValueError('Server config entry "name" is missing') # encoding defaults to latin1 if "encoding" not in d: d["encoding"] = "latin1" # Generate URL entries from CGI entries, if cgi_baseurl exists and # the specific URL entry is not already set. if "cgi_baseurl" in d: cgi_baseurl = d["cgi_baseurl"] cgi_keys = [k for k in self._valid_keys if k.startswith("cgi") and k != "cgi_baseurl"] for k in cgi_keys: url_key = f"url{k[3:]}" cgi_value = d.get(k) if d.get(url_key) is None and cgi_value is not None: d[url_key] = urljoin(cgi_baseurl, cgi_value) def __repr__(self): return f'<ServerConfig: {self._d.get("name")}>' def __dir__(self): return dir(type(self)) + list(self.__dict__.keys()) + self._valid_keys def __getattr__(self, name): if name in self._valid_keys: return self._d.get(name) else: return object.__getattribute__(self, name) def __setattr__(self, name, value): if name in self._valid_keys: if not isinstance(value, str): raise ValueError(f"{name} config value must be a string") self._d[name] = value else: object.__setattr__(self, name, value)
[docs] def copy(self): return ServerConfig(self._d)
[docs] def to_dict(self): return self._d
[docs] def check_supported(self, op): """ Check if an operation is supported by the server. """ if op == "series": return (self.cgi_show_series is not None) or (self.cgi_show_series_wrapper is not None) elif op == "info": return self.cgi_jsoc_info is not None elif op == "query": return self.cgi_jsoc_info is not None elif op == "email": return self.cgi_check_address is not None elif op == "export": return (self.cgi_jsoc_info is not None) and (self.cgi_jsoc_fetch is not None) else: raise ValueError(f"Unknown operation: {op!r}")
[docs]def register_server(config): """ Register a server configuration. """ global _server_configs name = if name in _server_configs: raise RuntimeError(f"ServerConfig {name} already registered") _server_configs[] = config
# Registered servers _server_configs = {} # Register public JSOC DRMS server. register_server( ServerConfig( name="JSOC", cgi_baseurl="", cgi_show_series="show_series", cgi_jsoc_info="jsoc_info", cgi_jsoc_fetch="jsoc_fetch", cgi_check_address="", cgi_show_series_wrapper="showextseries", show_series_wrapper_dbhost="hmidb2", http_download_baseurl="", ftp_download_baseurl="", ) ) # Register KIS DRMS server. register_server( ServerConfig( name="KIS", cgi_baseurl="", cgi_show_series="show_series", cgi_jsoc_info="jsoc_info", ) )