Writes configuration data to an env file, preserving schema order if provided.
(
env_data: Dict[str, Any],
schema: Optional[List[Dict[str, Any]]] = None,
file_path: str = str(ENV_FILE),
)
| 51 | |
| 52 | |
| 53 | def write_env( |
| 54 | env_data: Dict[str, Any], |
| 55 | schema: Optional[List[Dict[str, Any]]] = None, |
| 56 | file_path: str = str(ENV_FILE), |
| 57 | ) -> None: |
| 58 | """Writes configuration data to an env file, preserving schema order if provided.""" |
| 59 | log.debug("Writing env file", hypothesisId="CFG", path=file_path) |
| 60 | # Only load from file, do NOT include os.environ to avoid polluting env_file |
| 61 | original_config = load_env(file_path, include_os_environ=False) |
| 62 | lines_to_write: List[str] = [] |
| 63 | |
| 64 | schema_keys = set() |
| 65 | if schema: |
| 66 | schema_keys = {item["name"] for item in schema} |
| 67 | for item in schema: |
| 68 | key = item["name"] |
| 69 | value = env_data.get(key, item.get("default", "")) |
| 70 | if isinstance(value, list): |
| 71 | value = ",".join(value) |
| 72 | value_str = str(value).strip().replace("\n", " ").replace("\r", "") |
| 73 | lines_to_write.append(f"{key}={value_str}\n\n") |
| 74 | |
| 75 | # Append non-schema variables |
| 76 | header_added = False |
| 77 | for key, value in original_config.items(): |
| 78 | if key not in schema_keys: |
| 79 | if not header_added and schema: |
| 80 | lines_to_write.append("\n") |
| 81 | header_added = True |
| 82 | lines_to_write.append(f"{key}={value}\n\n") |
| 83 | |
| 84 | try: |
| 85 | with open(file_path, "w", newline="", encoding="utf-8") as f: |
| 86 | f.writelines(lines_to_write) |
| 87 | except Exception as e: |
| 88 | log.error(f"Error writing {file_path}: {e}", hypothesisId="CFG") |
| 89 | raise e |
| 90 | |
| 91 | |
| 92 | def env_flag(name: str, default: bool = False) -> bool: |
nothing calls this directly
no test coverage detected