Coverage for src/lilbee/server/handlers/config.py: 100%

1"""Config read/update handlers for the HTTP server.""" 

2 

3from __future__ import annotations 

4 

5import copy 

6import functools 

7from typing import Any 

8 

9from pydantic_core import PydanticUndefined 

10 

11from lilbee.app.settings import ( 

12 apply_settings_update, 

13 provider_reset_refused_message, 

14 requires_services_reset, 

15) 

16from lilbee.config_meta import ( 

17 MODEL_ROLE_FIELDS as _MODEL_ROLE_FIELDS, 

18) 

19from lilbee.config_meta import ( 

20 PUBLIC_CONFIG_FIELDS as _PUBLIC_CONFIG_FIELDS, 

21) 

22from lilbee.config_meta import ( 

23 WRITABLE_CONFIG_FIELDS, 

24) 

25from lilbee.core.config import Config, cfg 

26from lilbee.server.models import ConfigResponse, ConfigUpdateResponse 

27 

28 

29async def update_config(updates: dict[str, Any]) -> ConfigUpdateResponse: 

30 """Partial update of writable config fields. 

31 

32 Delegates validation, snapshot/rollback, persistence, and cache 

33 invalidation to ``app.settings.apply_settings_update`` so HTTP, MCP, 

34 CLI, and the TUI share one write boundary. Model role writes are 

35 refused at this surface because PUT /api/models/<role> already 

36 handles them with an install-availability check. 

37 

38 A provider switch rebuilds the shared Services singleton, unsafe while other 

39 clients have in-flight requests, so it is refused on the always-concurrent 

40 HTTP server; do it from the CLI instead. 

41 """ 

42 if requires_services_reset(updates): 

43 raise ValueError(provider_reset_refused_message("Switching")) 

44 result = apply_settings_update(updates, allow_model_roles=False) 

45 return ConfigUpdateResponse(updated=result.updated, reindex_required=result.reindex_required) 

46 

47 

48async def get_config() -> ConfigResponse: 

49 """Return all user-facing configuration values.""" 

50 dumped = cfg.model_dump() 

51 result = {k: v for k, v in dumped.items() if k in _PUBLIC_CONFIG_FIELDS} 

52 return ConfigResponse(**result) 

53 

54 

55@functools.cache 

56def _compute_config_defaults() -> dict[str, Any]: 

57 """Materialize Config defaults once per process.""" 

58 defaults: dict[str, Any] = {} 

59 for name, info in Config.model_fields.items(): 

60 # The writable conjunct is redundant today (every public field is 

61 # writable or a model role) but keeps a future public-but-not-writable 

62 # field out of this payload. 

63 is_writable_public = name in WRITABLE_CONFIG_FIELDS and name in _PUBLIC_CONFIG_FIELDS 

64 if not is_writable_public and name not in _MODEL_ROLE_FIELDS: 

65 continue 

66 value = info.get_default(call_default_factory=True) 

67 if value is PydanticUndefined: # pragma: no cover 

68 continue 

69 defaults[name] = value 

70 return defaults 

71 

72 

73async def get_config_defaults() -> ConfigResponse: 

74 """Return canonical defaults for every public config field. 

75 

76 Covers writable fields (resettable via PATCH /api/config) and the 

77 model-role fields (resettable via PUT /api/models/<role>). 

78 

79 Deepcopies the cached dict so callers that mutate the response 

80 (list-valued fields like ``crawl_exclude_patterns``) cannot poison 

81 subsequent calls. 

82 """ 

83 return ConfigResponse(**copy.deepcopy(_compute_config_defaults()))