Coverage for arclith / infrastructure / config.py: 96%

1from __future__ import annotations 

2 

3from pathlib import Path 

4from typing import Literal 

5 

6import yaml 

7from pydantic import BaseModel, field_validator, model_validator 

8 

9_DUCKDB_SUPPORTED_EXTENSIONS = {".csv", ".parquet", ".json", ".arrow"} 

10 

11 

12class MongoDBSettings(BaseModel): 

13 uri: str | None = None 

14 db_name: str 

15 collection_name: str | None = None 

16 

17 

18class DuckDBSettings(BaseModel): 

19 path: str 

20 

21 @field_validator("path") 

22 @classmethod 

23 def must_be_supported_format(cls, v: str) -> str: 

24 p = Path(v) 

25 if p.is_dir() or v.endswith("/"): 

26 return v 

27 ext = p.suffix.lower() 

28 if ext not in _DUCKDB_SUPPORTED_EXTENSIONS: 

29 raise ValueError( 

30 f"Format '{ext}' non supporté par DuckDB. " 

31 f"Formats acceptés : {', '.join(sorted(_DUCKDB_SUPPORTED_EXTENSIONS))}" 

32 ) 

33 return v 

34 

35 

36class SoftDeleteSettings(BaseModel): 

37 retention_days: float | None = None 

38 

39 @field_validator("retention_days") 

40 @classmethod 

41 def must_be_positive(cls, v: float | None) -> float | None: 

42 if v is not None and v < 0: 

43 raise ValueError("retention_days doit être >= 0") 

44 return v 

45 

46 

47class AdaptersSettings(BaseModel): 

48 logger: Literal["console"] = "console" 

49 repository: Literal["memory", "mongodb", "duckdb"] = "memory" 

50 multitenant: bool = False 

51 mongodb: MongoDBSettings | None = None 

52 duckdb: DuckDBSettings | None = None 

53 

54 @model_validator(mode = "after") 

55 def validate_repository_config(self) -> "AdaptersSettings": 

56 if self.repository == "mongodb": 

57 if self.mongodb is None: 

58 raise ValueError("repository=mongodb mais aucune section [adapters.mongodb] dans config.yaml") 

59 elif self.repository == "duckdb" and self.duckdb is None: 

60 raise ValueError("repository=duckdb mais aucune section [adapters.duckdb] dans config.yaml") 

61 return self 

62 

63 

64class ApiSettings(BaseModel): 

65 host: str = "0.0.0.0" # nosec B104 

66 port: int = 8000 

67 reload: bool = True 

68 

69 

70class McpSettings(BaseModel): 

71 host: str = "127.0.0.1" 

72 port: int = 8001 

73 

74 

75class AppConfig(BaseModel): 

76 adapters: AdaptersSettings = AdaptersSettings() 

77 soft_delete: SoftDeleteSettings = SoftDeleteSettings() 

78 api: ApiSettings = ApiSettings() 

79 mcp: McpSettings = McpSettings() 

80 

81 

82def load_config(path: Path) -> AppConfig: 

83 with open(path) as f: 

84 data = yaml.safe_load(f) 

85 data = data or {} 

86 

87 from arclith.infrastructure.secret_factory import build_secret_resolver 

88 from arclith.infrastructure.secret_loader import resolve_dict_secrets 

89 from contextlib import suppress 

90 

91 resolver = build_secret_resolver(data, path.parent) 

92 if resolver: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 with suppress(Exception): 

94 data = resolve_dict_secrets(data, resolver) 

95 

96 return AppConfig.model_validate(data)