Coverage for src/cstlcore/settings.py: 94%

1import logging 

2from enum import Enum 

3from functools import cached_property 

4from pathlib import Path 

5from typing import ClassVar 

6 

7from pydantic import ( 

8 BaseModel, 

9 Field, 

10 HttpUrl, 

11 PostgresDsn, 

12 ValidationError, 

13 model_validator 

14) 

15from pydantic_settings import BaseSettings, SettingsConfigDict 

16from sqlalchemy.engine.url import URL 

17 

18 

19class NewsletterConfig(BaseModel): 

20 secret_key: str | None = None 

21 # Temporary, should be removed later 

22 admin_token: str | None = None 

23 

24 @model_validator(mode="after") 

25 def validate_secret_key(self) -> "NewsletterConfig": 

26 if not self.secret_key: 

27 logging.warning( 

28 "[Settings] NEWSLETTER__SECRET_KEY not set. Newsletter subscription disabled." 

29 ) 

30 return self 

31 

32 

33class HostnameConfig(BaseModel): 

34 current: HttpUrl = Field(default=HttpUrl("http://localhost:8001")) 

35 graph_api: HttpUrl = Field(default=HttpUrl("http://localhost:8000")) 

36 sse: HttpUrl = Field(default=HttpUrl("http://localhost:8002")) 

37 frontend: HttpUrl = Field(default=HttpUrl("http://localhost:3000")) 

38 

39 

40class PostgresConfig(BaseModel): 

41 user: str = Field(default="postgres") 

42 password: str = Field(default="password", min_length=8) 

43 host: str = Field(default="localhost") 

44 port: int = Field(default=5432) 

45 db_name: str = Field(default="cstl", alias="db") 

46 test_db_name: str = Field(default="test", alias="test_db") 

47 test_port: int = Field(default=5433) 

48 

49 @cached_property 

50 def uri(self) -> URL: 

51 url = URL.create( 

52 "postgresql", 

53 username=self.user, 

54 password=self.password, 

55 host=self.host, 

56 port=self.port, 

57 database=self.db_name, 

58 ) 

59 return url 

60 

61 @cached_property 

62 def test_uri(self) -> URL: 

63 url = URL.create( 

64 "postgresql", 

65 username=self.user, 

66 password=self.password, 

67 host=self.host, 

68 port=self.test_port, 

69 database=self.test_db_name, 

70 ) 

71 return url 

72 

73 @model_validator(mode="after") 

74 def validate_uri(self) -> "PostgresConfig": 

75 # Validate the URI once after model initialization 

76 try: 

77 _ = PostgresDsn(str(self.uri)) 

78 except ValidationError as e: 

79 raise ValueError(f"Invalid Postgres URI: {self.uri}") from e 

80 return self 

81 

82 

83class GoogleAuthConfig(BaseModel): 

84 client_id: str | None = None 

85 client_secret: str | None = None 

86 redirect_uri: HttpUrl = HttpUrl("http://localhost:8001/auth/google/callback") 

87 token_uri: HttpUrl = HttpUrl("https://oauth2.googleapis.com/token") 

88 auth_uri: HttpUrl = HttpUrl("https://accounts.google.com/o/oauth2/auth") 

89 

90 @model_validator(mode="after") 

91 def validate_client_credentials(self) -> "GoogleAuthConfig": 

92 if not self.client_id or not self.client_secret: 

93 logging.warning( 

94 "[Settings] Google OAuth client_id or client_secret not set. Google authentication disabled." 

95 ) 

96 return self 

97 

98 

99class SMTPConfig(BaseModel): 

100 email: str | None = None 

101 password: str | None = None 

102 

103 @model_validator(mode="after") 

104 def validate_smtp_credentials(self) -> "SMTPConfig": 

105 if not self.email or not self.password: 

106 logging.warning( 

107 "[Settings] SMTP email or password not set. Email sending disabled." 

108 ) 

109 return self 

110 

111 

112class JWTConfig(BaseModel): 

113 secret_key: str | None = None 

114 algorithm: str = Field(default="HS256") 

115 access_token_expire_hours: int = Field(default=24) 

116 

117 @model_validator(mode="after") 

118 def validate_secret_key(self) -> "JWTConfig": 

119 if not self.secret_key: 

120 raise ValueError(f"JWT__SECRET_KEY must be set, current: {self.secret_key}") 

121 return self 

122 

123 

124class FileSystemConfig(BaseModel): 

125 data_directory: Path = Field(default=Path("./data")) 

126 map_directory: Path = Field(default=Path("./data/maps")) 

127 

128 def ensure_directories(self) -> None: 

129 for path in [self.data_directory, self.map_directory]: 

130 if not path.exists(): 

131 try: 

132 path.mkdir(parents=True, exist_ok=True) 

133 except Exception as e: 

134 raise RuntimeError(f"Failed to create directory {path}: {e}") 

135 

136 

137class Target(str, Enum): 

138 dev = "dev" 

139 prod = "prod" 

140 test = "test" 

141 

142class AdminConfig(BaseModel): 

143 emails: str | list[str] = Field( 

144 default="", 

145 description="Comma-separated list of admin email addresses" 

146 ) 

147 

148 @model_validator(mode="after") 

149 def parse_emails(self) -> "AdminConfig": 

150 if self.emails and isinstance(self.emails, str): 

151 self.emails = [email.strip() for email in self.emails.split(",") if email.strip()] 

152 else: 

153 self.emails = [] 

154 return self 

155 

156 def is_admin_email(self, email: str) -> bool: 

157 """Check if an email is in the admin list (case-insensitive)""" 

158 return email.lower() in [admin_email.lower() for admin_email in self.emails] 

159 

160 

161class Settings(BaseSettings): 

162 model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict( 

163 env_file=".env", 

164 env_file_encoding="utf-8", 

165 extra="ignore", 

166 env_nested_delimiter="__", 

167 ) 

168 

169 target: Target = Field(default=Target.dev) 

170 allowed_origins: list[str] = Field( 

171 default=[ 

172 "http://localhost:3000", 

173 "http://localhost:8000", 

174 "http://localhost:8001", 

175 ] 

176 ) 

177 landing_page_url: HttpUrl = Field(default=HttpUrl("http://localhost:3000/login")) 

178 

179 fs: FileSystemConfig = Field(default_factory=FileSystemConfig) 

180 google_auth: GoogleAuthConfig = Field(default_factory=GoogleAuthConfig) 

181 postgres: PostgresConfig = Field(default_factory=PostgresConfig) 

182 jwt: JWTConfig = Field(default_factory=JWTConfig) 

183 newsletter: NewsletterConfig = Field(default_factory=NewsletterConfig) 

184 smtp: SMTPConfig = Field(default_factory=SMTPConfig) 

185 admin: AdminConfig = Field(default_factory=AdminConfig) 

186 

187 # Important, when using docker each will see an environment variable 

188 # named HOSTNAME such as HOSTNAME=8e2b7c3e1c7a 

189 services: HostnameConfig = Field(default_factory=HostnameConfig, alias="SERVICE") 

190 

191 @model_validator(mode="after") 

192 def validate_allowed_origins(self) -> "Settings": 

193 for origin in self.allowed_origins: 

194 try: 

195 _ = HttpUrl(origin) 

196 except ValidationError as e: 

197 raise ValueError(f"Invalid allowed_origin: {origin}") from e 

198 return self 

199 

200 @model_validator(mode="after") 

201 def set_cryptographic_keys(self) -> "Settings": 

202 if not self.newsletter.secret_key: 

203 self.newsletter.secret_key = self.jwt.secret_key 

204 return self 

205 

206 

207settings = Settings() 

208settings.fs.ensure_directories()