Coverage for mindsdb / integrations / handlers / mysql_handler / settings.py: 47%

1from typing import Optional 

2from pydantic import BaseModel, AnyUrl, TypeAdapter, model_validator, field_validator, ConfigDict 

3from urllib.parse import urlparse 

4 

5 

6_ANY_URL_ADAPTER = TypeAdapter(AnyUrl) 

7 

8 

9class ConnectionConfig(BaseModel): 

10 """ 

11 MySQL connection configuration with validation. 

12 

13 Supports two connection methods: 

14 1. URL-based: mysql://user:password@host:port/database 

15 2. Parameter-based: individual host, port, user, password, database params 

16 """ 

17 

18 url: Optional[AnyUrl] = None 

19 host: Optional[str] = None 

20 port: int = 3306 

21 user: Optional[str] = None 

22 password: Optional[str] = None 

23 database: Optional[str] = None 

24 

25 @field_validator("port") 

26 @classmethod 

27 def validate_port(cls, v: int) -> int: 

28 """Validate that port is within valid range.""" 

29 if v < 1 or v > 65535: 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true

30 raise ValueError(f"Port must be between 1 and 65535, got {v}") 

31 return v 

32 

33 @field_validator("url", mode="before") 

34 @classmethod 

35 def validate_url(cls, v: Optional[str]) -> Optional[AnyUrl]: 

36 """Validate URL using AnyUrl as a fallback option for MySQL DSN parsing.""" 

37 if v is None or isinstance(v, AnyUrl): 

38 return v 

39 try: 

40 return _ANY_URL_ADAPTER.validate_python(v) 

41 except ValueError as exc: 

42 raise ValueError(f"Invalid MySQL connection URL: {v}") from exc 

43 

44 @field_validator("host") 

45 @classmethod 

46 def validate_host(cls, v: Optional[str]) -> Optional[str]: 

47 """Validate that host is not empty if provided.""" 

48 if v is not None and not v.strip(): 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 raise ValueError("Host cannot be empty string") 

50 return v 

51 

52 @field_validator("database") 

53 @classmethod 

54 def validate_database(cls, v: Optional[str]) -> Optional[str]: 

55 """Validate that database name is not empty if provided.""" 

56 if v is not None and not v.strip(): 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 raise ValueError("Database name cannot be empty string") 

58 return v 

59 

60 @model_validator(mode="before") 

61 @classmethod 

62 def check_db_params(cls, values): 

63 """Ensures either URL is provided or all individual parameters are provided.""" 

64 url = values.get("url") 

65 host = values.get("host") 

66 user = values.get("user") 

67 password = values.get("password") 

68 database = values.get("database") 

69 

70 if not url and not (host and user and password and database): 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 missing_params = [] 

72 if not host: 

73 missing_params.append("host") 

74 if not user: 

75 missing_params.append("user") 

76 if not password: 

77 missing_params.append("password") 

78 if not database: 

79 missing_params.append("database") 

80 

81 raise ValueError( 

82 f"Either a valid URL or all required parameters must be provided. Missing: {', '.join(missing_params)}" 

83 ) 

84 

85 if url: 85 ↛ 87line 85 didn't jump to line 87 because the condition on line 85 was never true

86 # Parse URL and extract connection parameters 

87 try: 

88 parsed = urlparse(str(url)) 

89 

90 # Extract parameters from URL 

91 values["host"] = parsed.hostname or host 

92 values["port"] = parsed.port if parsed.port is not None else values.get("port", 3306) 

93 values["user"] = parsed.username or user 

94 values["password"] = parsed.password or password 

95 values["database"] = parsed.path[1:] if parsed.path and len(parsed.path) > 1 else database 

96 

97 # Validate extracted parameters 

98 if not values["host"]: 

99 raise ValueError("URL must contain a hostname") 

100 if not values["user"]: 

101 raise ValueError("URL must contain a username") 

102 if not values["database"]: 

103 raise ValueError("URL must contain a database name in the path") 

104 

105 except Exception as e: 

106 raise ValueError(f"Invalid MySQL connection URL: {str(e)}") 

107 

108 # mysql connector raises error if url is provided 

109 values.pop("url", None) 

110 

111 return values 

112 

113 # Validate individual parameters 

114 if not url: 114 ↛ 119line 114 didn't jump to line 119 because the condition on line 114 was always true

115 for param in ["host", "user", "password", "database"]: 

116 if not values.get(param): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true

117 raise ValueError(f"'{param}' is required when URL is not provided") 

118 

119 return values 

120 

121 model_config = ConfigDict(str_min_length=1, str_strip_whitespace=True, validate_assignment=True)