Coverage for snekql/mariadb/config.py: 89%

1"""MariaDB runtime configuration for snekql.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from pathlib import Path 

7 

8from snekql.errors import DatabaseRuntimeError 

9from snekql.validation import NonNegativeFloat, PositiveInt, validate_boundary 

10 

11_MAX_TCP_PORT = 65535 

12 

13 

14@validate_boundary(error_type=DatabaseRuntimeError) 

15def _validate_numeric_config( 

16 *, 

17 acquire_timeout: NonNegativeFloat, 

18 pool_size: PositiveInt, 

19 port: PositiveInt, 

20) -> None: 

21 """Validate numeric settings before semantic connection checks run.""" 

22 

23 del acquire_timeout, pool_size, port 

24 

25 

26def _validate_non_empty_string(name: str, value: str) -> None: 

27 """Reject empty string settings that cannot identify a database endpoint.""" 

28 

29 if value.strip() == "": 

30 msg = f"MariaDB {name} must not be empty" 

31 raise DatabaseRuntimeError(msg) 

32 

33 

34def _validate_port(port: PositiveInt) -> None: 

35 """MariaDB TCP ports must fit the valid TCP port range.""" 

36 

37 if port > _MAX_TCP_PORT: 

38 msg = "MariaDB port must be between 1 and 65535" 

39 raise DatabaseRuntimeError(msg) 

40 

41 

42@dataclass(frozen=True, kw_only=True) 

43class Config: 

44 """MariaDB backend configuration for explicit runtime initialization. 

45 

46 >>> config = Config(database="app", user="snekql") 

47 >>> config.port 

48 3306 

49 """ 

50 

51 database: str 

52 acquire_timeout: NonNegativeFloat = 30.0 

53 charset: str = "utf8mb4" 

54 host: str = "127.0.0.1" 

55 password: str = field(default="", repr=False) 

56 pool_size: PositiveInt = 5 

57 port: PositiveInt = 3306 

58 unix_socket: Path | None = None 

59 user: str 

60 

61 def __post_init__(self) -> None: 

62 _validate_numeric_config( 

63 acquire_timeout=self.acquire_timeout, 

64 pool_size=self.pool_size, 

65 port=self.port, 

66 ) 

67 _validate_port(self.port) 

68 _validate_non_empty_string("database", self.database) 

69 _validate_non_empty_string("host", self.host) 

70 _validate_non_empty_string("user", self.user) 

71 _validate_non_empty_string("charset", self.charset)