Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/config_resolvers/_descriptors.py: 84%

1"""Shared primitive for resolving a settings value from its sources. 

2 

3A ``FieldDescriptor`` names, for one setting, where it lives in the stored DB 

4row (``db_key``), which process env var carries it (``env_var``), whether it is 

5a secret, and its effective default. ``resolve_fields`` reconciles a set of 

6descriptors against a decrypted DB row and the process environment with a fixed 

7precedence, returning the resolved values plus per-field provenance so a caller 

8can tell whether a value came from the database, the environment, a default, or 

9is unset. 

10""" 

11 

12from collections.abc import Mapping, Sequence 

13from dataclasses import dataclass 

14from typing import Final, Literal 

15 

16FieldSource = Literal["config", "db", "env", "default", "unset"] 

17 

18 

19@dataclass(frozen=True, slots=True) 

20class FieldDescriptor: 

21 field_name: str 

22 db_key: str 

23 env_var: str 

24 is_secret: bool = False 

25 default: str | None = None 

26 

27 

28def _db_is_set(db_value: object, empty_db_is_set: bool) -> bool: 

29 if empty_db_is_set: 29 ↛ 32line 29 didn't jump to line 32 because the condition on line 29 was never true

30 # A stored key that is present, even as "", is an explicit admin choice 

31 # (e.g. clearing an alerting webhook) and must win over a stale env var. 

32 return db_value is not None 

33 # A blank stored value is treated as absent, so it falls through to env. This 

34 # fits settings whose clear path also unsets the env var (e.g. SSO). 

35 return isinstance(db_value, str) and bool(db_value.strip()) 

36 

37 

38def _resolve_one( 

39 descriptor: FieldDescriptor, 

40 db_values: Mapping[str, object], 

41 env: Mapping[str, str], 

42 empty_db_is_set: bool, 

43) -> tuple[str, str | None, FieldSource]: 

44 db_value: Final = db_values.get(descriptor.db_key) 

45 if _db_is_set(db_value, empty_db_is_set): 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 return descriptor.field_name, db_value if isinstance(db_value, str) else str(db_value), "db" 

47 env_value: Final = env.get(descriptor.env_var) 

48 if isinstance(env_value, str) and env_value.strip(): 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 return descriptor.field_name, env_value, "env" 

50 if descriptor.default is not None: 

51 return descriptor.field_name, descriptor.default, "default" 

52 return descriptor.field_name, None, "unset" 

53 

54 

55def resolve_fields( 

56 descriptors: Sequence[FieldDescriptor], 

57 db_values: Mapping[str, object], 

58 env: Mapping[str, str], 

59 empty_db_is_set: bool = False, 

60) -> tuple[dict[str, str | None], dict[str, FieldSource]]: 

61 """Resolve every descriptor to (values, provenance). 

62 

63 Precedence per field: a set stored value wins, else a non-blank process env 

64 var, else the descriptor default, else unset. ``empty_db_is_set`` selects 

65 how a present-but-empty stored value is read: ``False`` treats it as absent 

66 so it falls back to env (SSO, whose clear path also unsets the env var); 

67 ``True`` treats it as an explicit clear that wins over env (alerting, whose 

68 clear path stores "" without unsetting the env var). 

69 """ 

70 resolved: Final = tuple(_resolve_one(descriptor, db_values, env, empty_db_is_set) for descriptor in descriptors) 

71 values: Final = {field_name: value for field_name, value, _ in resolved} 

72 provenance: Final[dict[str, FieldSource]] = dict( # mutable-ok: public resolver contract returns a plain dict 

73 (field_name, source) for field_name, _, source in resolved 

74 ) 

75 return values, provenance