sunnylink: enhanced param keys fetch with data type (#1308)
* refactor: enhance parameter handling and add new parameter retrieval method - Refactored `get_param_as_byte` for better modularity and added `_to_bytes` helper function. - Introduced `getParamsAllKeysV1` to retrieve all keys with enhanced metadata. * refactor: update parameter handling and response structure in sunnylink - Modified `_to_bytes` to accept `bytes` type for improved type consistency. - Adjusted response keys in `sunnylinkd` for clarity. * fix: update `get_param_as_byte` to use corrected method for default values - Replaced `get_default` with `get_default_value` for accurate param retrieval. - Ensures consistent handling of default parameter values. * refactor: remove redundant `None` check in `sunnylinkd.py` - Streamlined parameter iteration by eliminating unnecessary `None` validation. - Simplifies logic for constructing `params_dict`. * refactor: streamline `sunnylinkd` response by removing redundant `keys_v1` field - Simplified return structure by excluding unused `keys_v1` key. - Reduces response payload for improved efficiency. * refactor: split `getParamsAllKeys` into distinct functions for improved clarity - Added `getParamsAllKeysV1` to preserve original metadata-rich behavior. - Revised `getParamsAllKeys` to return a simplified list of parameter keys. * style: remove extraneous newline in `sunnylinkd.py` - Ensures consistent formatting and adheres to style guidelines. * Test
This commit is contained in:
@@ -178,12 +178,29 @@ def getParamsAllKeys() -> list[str]:
|
||||
return keys
|
||||
|
||||
|
||||
@dispatcher.add_method
|
||||
def getParamsAllKeysV1() -> dict[str, str]:
|
||||
available_keys: list[str] = [k.decode('utf-8') for k in Params().all_keys()]
|
||||
|
||||
params_dict: dict[str, list[dict[str, str | bool | int | None]]] = {"params": []}
|
||||
for key in available_keys:
|
||||
value = get_param_as_byte(key, get_default=True)
|
||||
params_dict["params"].append({
|
||||
"key": key,
|
||||
"type": int(params.get_type(key).value),
|
||||
"default_value": base64.b64encode(value).decode('utf-8') if value else None,
|
||||
})
|
||||
|
||||
return {"keys": json.dumps(params_dict.get("params", []))}
|
||||
|
||||
|
||||
@dispatcher.add_method
|
||||
def getParams(params_keys: list[str], compression: bool = False) -> str | dict[str, str]:
|
||||
params = Params()
|
||||
available_keys: list[str] = [k.decode('utf-8') for k in Params().all_keys()]
|
||||
|
||||
try:
|
||||
param_keys_validated = [key for key in params_keys if key in getParamsAllKeys()]
|
||||
param_keys_validated = [key for key in params_keys if key in available_keys]
|
||||
params_dict: dict[str, list[dict[str, str | bool | int]]] = {"params": []}
|
||||
for key in param_keys_validated:
|
||||
value = get_param_as_byte(key)
|
||||
|
||||
@@ -60,21 +60,25 @@ def get_api_token():
|
||||
print(f"API Token: {token}")
|
||||
|
||||
|
||||
def get_param_as_byte(param_name: str, params=None) -> bytes | None:
|
||||
"""Get a parameter as bytes. Returns None if the parameter does not exist."""
|
||||
params = params or Params() # Use existing Params instance if provided
|
||||
param = params.get(param_name)
|
||||
if param is None:
|
||||
def _to_bytes(value: bytes | None, param_type: ParamKeyType) -> bytes | None:
|
||||
"""Convert a parameter value to bytes based on its type."""
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
param_type = params.get_type(param_name)
|
||||
if param_type == ParamKeyType.BYTES:
|
||||
return bytes(param)
|
||||
elif param_type == ParamKeyType.JSON:
|
||||
return json.dumps(param).encode('utf-8')
|
||||
return str(param).encode('utf-8')
|
||||
return bytes(value)
|
||||
if param_type == ParamKeyType.JSON:
|
||||
return json.dumps(value).encode("utf-8")
|
||||
return str(value).encode("utf-8")
|
||||
|
||||
|
||||
def get_param_as_byte(param_name: str, params=None, get_default=False) -> bytes | None:
|
||||
"""Get a parameter as bytes. Returns None if the parameter does not exist."""
|
||||
params = params or Params()
|
||||
value = params.get(param_name) if not get_default else params.get_default_value(param_name)
|
||||
param_type = params.get_type(param_name)
|
||||
return _to_bytes(value, param_type)
|
||||
|
||||
def save_param_from_base64_encoded_string(param_name: str, base64_encoded_data: str, is_compressed=False) -> None:
|
||||
"""Save a parameter from bytes. Overwrites the parameter if it already exists."""
|
||||
params = Params()
|
||||
|
||||
Reference in New Issue
Block a user