|
1 | 1 | import yaml |
2 | 2 | from pathlib import Path |
| 3 | +from types import SimpleNamespace |
| 4 | +from typing import Any, Optional, Union, Dict |
3 | 5 |
|
4 | 6 |
|
5 | | -class YamlReader: |
| 7 | +class YAMLReader: |
| 8 | + """ |
| 9 | + Reads data from a YAML file and returns it in the specified format. |
| 10 | +
|
| 11 | + :param filename (str): The name of the YAML file to read |
| 12 | + :param to_simple_namespace (bool): If True, |
| 13 | + converts the returned data to SimpleNamespace. |
| 14 | + :param is_secure (bool): If True, |
| 15 | + use safe loading of YAML and decrypts passwords. |
| 16 | + :param array_of_all_values (bool): If True, |
| 17 | + return all values in a flattened dictionary. |
| 18 | + :param separator (Optional[str]): String used to separate keys in the |
| 19 | + flattened dictionary. |
| 20 | +
|
| 21 | + Returns: |
| 22 | + Union[SimpleNamespace, dict, list]: The parsed data from the YAML |
| 23 | + which can be a SimpleNamespace, dictionary, or list |
| 24 | + """ |
| 25 | + |
6 | 26 | @staticmethod |
7 | | - def read_caps(browser="chrome", filename="caps.yaml"): |
| 27 | + def read( |
| 28 | + filename: str = "data.yaml", |
| 29 | + to_simple_namespace: bool = False, |
| 30 | + is_secure: bool = False, |
| 31 | + array_of_all_values: bool = False, |
| 32 | + separator: Optional[str] = None, |
| 33 | + ) -> Union[SimpleNamespace, dict, list]: |
| 34 | + resources_path = Path(__file__).resolve().parent.parent / "config" |
| 35 | + abs_path = resources_path / filename |
| 36 | + |
| 37 | + if not abs_path.exists(): |
| 38 | + raise FileNotFoundError(f"The file {abs_path} does not exist.") |
| 39 | + |
8 | 40 | try: |
9 | | - # Get the path to the resources folder from the script's directory |
10 | | - resources_path = ( |
11 | | - Path(__file__).resolve().parent.parent.parent / "config" |
12 | | - ) |
| 41 | + with open(abs_path, "r", encoding="UTF-8") as stream: |
| 42 | + if is_secure: |
| 43 | + data = yaml.safe_load(stream) |
| 44 | + data = YAMLReader._decrypt_password( |
| 45 | + data |
| 46 | + ) # Decrypt passwords if necessary |
| 47 | + else: |
| 48 | + data = yaml.load(stream, Loader=yaml.FullLoader) |
| 49 | + |
| 50 | + except yaml.YAMLError as e: |
| 51 | + raise ValueError(f"Error loading YAML file: {e}") |
| 52 | + |
| 53 | + # Convert to SimpleNamespace |
| 54 | + if to_simple_namespace: |
| 55 | + data = YAMLReader._convert_to_namespace(data) |
| 56 | + |
| 57 | + # If array_of_all_values is True, return all values in a dictionary |
| 58 | + if array_of_all_values: |
| 59 | + return YAMLReader._flatten_values(data, separator) |
| 60 | + |
| 61 | + return data |
| 62 | + |
| 63 | + @staticmethod |
| 64 | + def _decrypt_password(data: Any) -> Any: |
| 65 | + """ |
| 66 | + Decrypts passwords in the data if they are encrypted. |
| 67 | + """ |
| 68 | + pass |
| 69 | + for key, value in data.items(): |
| 70 | + if key == "password": |
| 71 | + data[key] = YAMLReader._decrypt(value) |
| 72 | + # else: |
| 73 | + # data[key] = YAMLReader._decrypt_password(value) |
| 74 | + |
| 75 | + @staticmethod |
| 76 | + def _decrypt(encrypted_value: str) -> str: |
| 77 | + """ |
| 78 | + Replace with your actual decryption logic. |
| 79 | + you could use a library like `cryptography` or `PyCryptodome`. |
| 80 | + """ |
| 81 | + # Placeholder for decryption logic, modify as per your requirements |
| 82 | + return encrypted_value # Replace with actual decrypted value |
| 83 | + |
| 84 | + @staticmethod |
| 85 | + def read_caps( |
| 86 | + browser: str = "chrome", filename: str = "data.yaml" |
| 87 | + ) -> Optional[Dict[str, Any]]: |
| 88 | + """Read browser capabilities from a YAML file.""" |
| 89 | + try: |
| 90 | + resources_path = Path(__file__).resolve().parent.parent / "config" |
13 | 91 | abs_path = resources_path / filename |
14 | 92 |
|
15 | 93 | with open(abs_path, "r", encoding="UTF-8") as stream: |
16 | 94 | data = yaml.safe_load(stream) |
17 | | - return data.get(browser) |
| 95 | + return data.get( |
| 96 | + browser |
| 97 | + ) # Return capabilities for the specified browser |
18 | 98 | except (yaml.YAMLError, KeyError) as e: |
19 | 99 | print(f"Error while reading '{filename}': {e}") |
20 | 100 | return None |
| 101 | + |
| 102 | + @staticmethod |
| 103 | + def _convert_to_namespace( |
| 104 | + data: Any, |
| 105 | + ) -> Union[SimpleNamespace, list[SimpleNamespace], Any]: |
| 106 | + """Convert a dictionary to a SimpleNamespace.""" |
| 107 | + if isinstance(data, dict): |
| 108 | + return SimpleNamespace( |
| 109 | + **{k: YAMLReader._convert_to_namespace(v) for k, v in data.items()} |
| 110 | + ) |
| 111 | + elif isinstance(data, list): |
| 112 | + return [YAMLReader._convert_to_namespace(item) for item in data] |
| 113 | + return data |
| 114 | + |
| 115 | + @staticmethod |
| 116 | + def _flatten_values( |
| 117 | + data: Any, separator: Optional[str] = None, parent_key: str = "" |
| 118 | + ) -> Dict[str, Any]: |
| 119 | + """ |
| 120 | + Flatten all values from a dictionary or list into a single dictionary |
| 121 | + """ |
| 122 | + items = {} |
| 123 | + if isinstance(data, dict): |
| 124 | + for key, value in data.items(): |
| 125 | + new_key = f"{parent_key}{separator}{key}" if parent_key else key |
| 126 | + if isinstance(value, dict) or isinstance(value, list): |
| 127 | + items.update( |
| 128 | + YAMLReader._flatten_values(value, separator, new_key) |
| 129 | + ) |
| 130 | + else: |
| 131 | + items[new_key] = value |
| 132 | + elif isinstance(data, list): |
| 133 | + for index, item in enumerate(data): |
| 134 | + new_key = ( |
| 135 | + f"{parent_key}{separator}{index}" if parent_key else str(index) |
| 136 | + ) |
| 137 | + items.update(YAMLReader._flatten_values(item, separator, new_key)) |
| 138 | + return items |
| 139 | + |
| 140 | + |
| 141 | +# Example usage |
| 142 | +caps = YAMLReader.read_caps("chrome", "caps.yaml") |
| 143 | +# Example usage simple namespace |
| 144 | +simple = YAMLReader.read("data.yaml", to_simple_namespace=True) |
| 145 | +print(simple.users.username1) |
0 commit comments