Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-workers/src/lexigram/ai/workers/config.py: 94%

1"""Worker configuration for the Lexigram framework.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING, ClassVar, cast 

7 

8from lexigram.ai.workers.constants import ENV_NESTED_DELIMITER, ENV_PREFIX 

9from lexigram.config.base import BaseConfig 

10from lexigram.validation import ConfigDict, Field 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.core.config import ConfigIssue, Environment 

14 

15 

16@dataclass(init=False) 

17class WorkersConfig(BaseConfig): 

18 """Configuration for AI background workers. 

19 

20 Loaded from the ``ai_workers:`` key in application.yaml, with environment 

21 variable overrides via ``LEX_AI_WORKERS__*`` prefix. 

22 """ 

23 

24 config_section: ClassVar[str] = "ai_workers" 

25 

26 model_config: ClassVar[ConfigDict] = cast( 

27 "ConfigDict", 

28 { 

29 "env_prefix": ENV_PREFIX, 

30 "env_nested_delimiter": ENV_NESTED_DELIMITER, 

31 "extra": "ignore", 

32 }, 

33 ) 

34 

35 enabled: bool = Field( 

36 default=True, description="Master on/off switch for all background workers" 

37 ) 

38 batch_embedding_concurrency: int = Field( 

39 default=3, 

40 ge=1, 

41 description="Concurrency level for batch embedding execution", 

42 ) 

43 document_ingestion_concurrency: int = Field( 

44 default=3, 

45 ge=1, 

46 description="Concurrency level for document parsing and chunking", 

47 ) 

48 enable_maintenance: bool = Field( 

49 default=True, 

50 description="Enable vector store and cache maintenance tasks", 

51 ) 

52 dlq_check_interval: int = Field( 

53 default=60, 

54 ge=1, 

55 description="Interval in seconds for DLQ recovery sweeps", 

56 ) 

57 

58 def validate_for_environment( 

59 self, 

60 env: Environment | None = None, 

61 ) -> list[ConfigIssue]: 

62 """Check config is safe for the target environment.""" 

63 return [] 

64 

65 

66__all__ = ["WorkersConfig"]