File size: 3,998 Bytes
037b69a 2c96ecb dc04eee 037b69a 2c96ecb 037b69a 2c96ecb dc04eee 037b69a 2c96ecb 037b69a 3cc3731 037b69a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Admin config value state and API response assembly."""
import os
from typing import Any
from .keys import pool_tokens, split_key_pool
from .manifest import (
FIELD_BY_KEY,
FIELDS,
SECTIONS,
ConfigFieldSpec,
ConfigOptionSpec,
)
from .sources import (
configured_env_files,
dotenv_values_from_file,
explicit_env_path,
is_locked_source,
repo_env_path,
template_values,
)
from .status import provider_config_status
MASKED_SECRET = "********"
ValueState = dict[str, dict[str, Any]]
def normalize_for_env(value: Any) -> str:
"""Normalize a submitted admin value for dotenv persistence."""
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def display_value(field: ConfigFieldSpec, value: str) -> str:
"""Return the Admin UI display value for a raw config value."""
if field.secret and value:
return MASKED_SECRET
return value
def load_value_state() -> ValueState:
"""Load effective admin field values and their sources."""
values = template_values()
sources = {key: "template" if key in values else "default" for key in FIELD_BY_KEY}
for source, path in configured_env_files():
file_values = dotenv_values_from_file(path)
for key, value in file_values.items():
if key in FIELD_BY_KEY:
values[key] = value
sources[key] = source
for key in FIELD_BY_KEY:
if key in os.environ:
values[key] = os.environ[key]
sources[key] = "process"
return {
key: {
"value": values.get(key, ""),
"source": sources.get(key, "default"),
}
for key in FIELD_BY_KEY
}
def load_config_response() -> dict[str, Any]:
"""Return manifest and current config values for the admin UI."""
state = load_value_state()
fields: list[dict[str, Any]] = []
for field in FIELDS:
entry = state[field.key]
source = entry["source"]
raw_value = entry["value"]
pool = (
{
"key_count": len(split_key_pool(raw_value)),
"keys": list(pool_tokens(raw_value)),
}
if field.pool_supported
else {}
)
fields.append(
{
"key": field.key,
"label": field.label,
"section": field.section_id,
"type": field.field_type,
"value": display_value(field, raw_value),
"configured": bool(str(raw_value).strip()),
"source": source,
"locked": is_locked_source(source),
"secret": field.secret,
"advanced": field.advanced,
"restart_required": field.restart_required,
"session_sensitive": field.session_sensitive,
"pool_supported": field.pool_supported,
"options": [
(
{"value": option.value, "label": option.label}
if isinstance(option, ConfigOptionSpec)
else {"value": option, "label": option}
)
for option in field.options
],
"description": field.description,
**pool,
}
)
return {
"sections": [
{
"id": section.section_id,
"label": section.label,
"description": section.description,
"advanced": section.advanced,
}
for section in SECTIONS
],
"fields": fields,
"paths": {
"managed": str(repo_env_path()),
"repo": str(repo_env_path()),
"explicit": str(explicit_env_path()) if explicit_env_path() else None,
},
"provider_status": provider_config_status(state),
}
|