Coverage for opt/mealie/lib/python3.12/site-packages/mealie/core/settings/settings.py: 85%

1import logging

2import os

3import secrets

4from datetime import UTC, datetime

5from pathlib import Path

6from typing import Annotated, Any, NamedTuple

8from dateutil.tz import tzlocal

9from pydantic import PlainSerializer, field_validator

10from pydantic_settings import BaseSettings, SettingsConfigDict

12from mealie.core.settings.themes import Theme

14from .db_providers import AbstractDBProvider, db_provider_factory

15from .static import PACKAGE_DIR

18class ScheduleTime(NamedTuple):

19 hour: int

20 minute: int

23class FeatureDetails(NamedTuple):

24 enabled: bool

25 """Indicates if the feature is enabled or not"""

26 description: str | None

27 """Short description describing why the feature is not ready"""

29 def __str__(self):

30 s = f"Enabled: {self.enabled}"

31 if not self.enabled and self.description: 31 ↛ 33line 31 didn't jump to line 33 because the condition on line 31 was always true

32 s += f"\nReason: {self.description}"

33 return s

36MaskedNoneString = Annotated[

37 str | None,

38 PlainSerializer(lambda x: None if x is None else "*****", return_type=str | None),

39]

40"""

41Custom serializer for sensitive settings. If the setting is None, then will serialize as null, otherwise,

42the secret will be serialized as '*****'

43"""

46def determine_secrets(data_dir: Path, secret: str, production: bool) -> str:

47 if not production: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 return "shh-secret-test-key"

50 secrets_file = data_dir.joinpath(secret)

51 if secrets_file.is_file(): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true

52 with open(secrets_file) as f:

53 return f.read()

54 else:

55 data_dir.mkdir(parents=True, exist_ok=True)

56 with open(secrets_file, "w") as f:

57 new_secret = secrets.token_hex(32)

58 f.write(new_secret)

59 return new_secret

62def get_secrets_dir() -> str | None:

63 """

64 Returns a directory to load secret settings from, or `None` if the secrets

65 directory does not exist or cannot be accessed.

66 """

67 # Avoid a circular import by importing here instead of at the file's top-level.

68 # get_logger -> AppSettings -> get_logger

69 from mealie.core.root_logger import get_logger

71 logger = get_logger()

73 secrets_dir = "/run/secrets"

75 # Check that the secrets directory exists.

76 if not os.path.exists(secrets_dir): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 logger.warning(f"Secrets directory '{secrets_dir}' does not exist")

78 return None

80 # Likewise, check we have permission to read from the secrets directory.

81 if not os.access(secrets_dir, os.R_OK): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 logger.warning(f"Secrets directory '{secrets_dir}' cannot be read from. Check permissions")

83 return None

85 # The secrets directory exists and can be accessed.

86 return secrets_dir

89class AppLoggingSettings(BaseSettings):

90 """

91 Subset of AppSettings to only access logging-related settings.

93 This is separated out from AppSettings to allow logging during construction

94 of AppSettings.

95 """

97 TESTING: bool = False

98 PRODUCTION: bool

100 LOG_CONFIG_OVERRIDE: Path | None = None

101 """path to custom logging configuration file"""

102

103 LOG_LEVEL: str = "info"

104 """corresponds to standard Python log levels"""

105

106

107class AppSettings(AppLoggingSettings):

108 theme: Theme = Theme()

109

110 BASE_URL: str = "http://localhost:8080"

111 """trailing slashes are trimmed (ex. `http://localhost:8080/` becomes ``http://localhost:8080`)"""

112

113 STATIC_FILES: str = str(PACKAGE_DIR / "frontend")

114 """path to static files directory (ex. `mealie/dist`)"""

115

116 IS_DEMO: bool = False

117

118 HOST_IP: str = "*"

119

120 API_HOST: str = "0.0.0.0"

121 API_PORT: int = 9000

122 API_DOCS: bool = True

123 TOKEN_TIME: int = 48

124 """time in hours"""

125

126 @field_validator("TOKEN_TIME")

127 @classmethod

128 def validate_token_time(cls, v: int) -> int:

129 if v < 1: 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 raise ValueError("TOKEN_TIME must be at least 1 hour")

131 # If TOKEN_TIME is unreasonably high (e.g. hundreds of years), JWT encoding

132 # can overflow, so we set the max to 10 years (87600 hours).

133 if v > 87600: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 raise ValueError("TOKEN_TIME is too high; maximum is 87600 hours (10 years)")

135 return v

136

137 SECRET: str

138 SESSION_SECRET: str

139

140 GIT_COMMIT_HASH: str = "unknown"

141

142 ALLOW_SIGNUP: bool = False

143 ALLOW_PASSWORD_LOGIN: bool = True

144

145 DAILY_SCHEDULE_TIME: str = "23:45"

146 """Local server time, in HH:MM format. See `DAILY_SCHEDULE_TIME_UTC` for the parsed UTC equivalent"""

147

148 @property

149 def logger(self) -> logging.Logger:

150 # Avoid a circular import by importing here instead of at the file's top-level.

151 # get_logger -> AppSettings -> get_logger

152 from mealie.core.root_logger import get_logger

153

154 return get_logger()

155

156 @property

157 def DAILY_SCHEDULE_TIME_UTC(self) -> ScheduleTime:

158 """The DAILY_SCHEDULE_TIME in UTC, parsed into hours and minutes"""

159

160 # parse DAILY_SCHEDULE_TIME into hours and minutes

161 try:

162 hour_str, minute_str = self.DAILY_SCHEDULE_TIME.split(":")

163 local_hour = int(hour_str)

164 local_minute = int(minute_str)

165 except ValueError:

166 local_hour = 23

167 local_minute = 45

168 self.logger.exception(

169 f"Unable to parse {self.DAILY_SCHEDULE_TIME=} as HH:MM; defaulting to {local_hour}:{local_minute}"

170 )

171

172 # DAILY_SCHEDULE_TIME is in local time, so we convert it to UTC

173 local_tz = tzlocal()

174 now = datetime.now(local_tz)

175 local_time = now.replace(hour=local_hour, minute=local_minute)

176 utc_time = local_time.astimezone(UTC)

177

178 self.logger.debug(f"Local time: {local_hour}:{local_minute} | UTC time: {utc_time.hour}:{utc_time.minute}")

179 return ScheduleTime(utc_time.hour, utc_time.minute)

180

181 # ===============================================

182 # Security Configuration

183

184 SECURITY_MAX_LOGIN_ATTEMPTS: int = 5

185 SECURITY_USER_LOCKOUT_TIME: int = 24

186 "time in hours"

187

188 @field_validator("BASE_URL")

189 @classmethod

190 def remove_trailing_slash(cls, v: str) -> str:

191 if v and v[-1] == "/": 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true

192 return v[:-1]

193

194 return v

195

196 @property

197 def DOCS_URL(self) -> str | None:

198 return "/docs" if self.API_DOCS else None

199

200 @property

201 def REDOC_URL(self) -> str | None:

202 return "/redoc" if self.API_DOCS else None

203

204 # ===============================================

205 # Database Configuration

206

207 DB_ENGINE: str = "sqlite" # Options: 'sqlite', 'postgres'

208 DB_PROVIDER: AbstractDBProvider | None = None

209

210 SQLITE_MIGRATE_JOURNAL_WAL: bool = False

211

212 @property

213 def DB_URL(self) -> str | None:

214 return self.DB_PROVIDER.db_url if self.DB_PROVIDER else None

215

216 @property

217 def DB_URL_PUBLIC(self) -> str | None:

218 return self.DB_PROVIDER.db_url_public if self.DB_PROVIDER else None

219

220 DEFAULT_GROUP: str = "Home"

221 DEFAULT_HOUSEHOLD: str = "Family"

222

223 _DEFAULT_EMAIL: str = "changeme@example.com"

224 """

225 This is the default email used for the first user created in the database. This is only used if no users

226 exist in the database. it should no longer be set by end users.

227 """

228 _DEFAULT_PASSWORD: str = "MyPassword"

229 """

230 This is the default password used for the first user created in the database. This is only used if no users

231 exist in the database. it should no longer be set by end users.

232 """

233

234 # ===============================================

235 # Email Configuration

236

237 SMTP_HOST: str | None = None

238 SMTP_PORT: str | None = "587"

239 SMTP_FROM_NAME: str | None = "Mealie"

240 SMTP_FROM_EMAIL: str | None = None

241 SMTP_USER: MaskedNoneString = None

242 SMTP_PASSWORD: MaskedNoneString = None

243 SMTP_AUTH_STRATEGY: str | None = "TLS" # Options: 'TLS', 'SSL', 'NONE'

244

245 @property

246 def SMTP_ENABLE(self) -> bool:

247 return self.SMTP_FEATURE.enabled

248

249 @property

250 def SMTP_FEATURE(self) -> FeatureDetails:

251 return AppSettings.validate_smtp(

252 self.SMTP_HOST,

253 self.SMTP_PORT,

254 self.SMTP_FROM_NAME,

255 self.SMTP_FROM_EMAIL,

256 self.SMTP_AUTH_STRATEGY,

257 self.SMTP_USER,

258 self.SMTP_PASSWORD,

259 )

260

261 @staticmethod

262 def validate_smtp(

263 host: str | None = None,

264 port: str | None = None,

265 from_name: str | None = None,

266 from_email: str | None = None,

267 strategy: str | None = None,

268 user: str | None = None,

269 password: str | None = None,

270 ) -> FeatureDetails:

271 """Validates all SMTP variables are set"""

272 description = None

273 required = {

274 "SMTP_HOST": host,

275 "SMTP_PORT": port,

276 "SMTP_FROM_NAME": from_name,

277 "SMTP_FROM_EMAIL": from_email,

278 "SMTP_AUTH_STRATEGY": strategy,

279 }

280 missing_values = [key for (key, value) in required.items() if value is None]

281 if missing_values: 281 ↛ 284line 281 didn't jump to line 284 because the condition on line 281 was always true

282 description = f"Missing required values for {missing_values}"

283

284 if strategy and strategy.upper() in {"TLS", "SSL"}: 284 ↛ 291line 284 didn't jump to line 291 because the condition on line 284 was always true

285 required["SMTP_USER"] = user

286 required["SMTP_PASSWORD"] = password

287 if not description: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true

288 missing_values = [key for (key, value) in required.items() if value is None]

289 description = f"Missing required values for {missing_values} because SMTP_AUTH_STRATEGY is not None"

290

291 not_none = "" not in required.values() and None not in required.values()

292

293 return FeatureDetails(enabled=not_none, description=description)

294

295 # ===============================================

296 # LDAP Configuration

297

298 LDAP_AUTH_ENABLED: bool = False

299 LDAP_SERVER_URL: str | None = None

300 LDAP_TLS_INSECURE: bool = False

301 LDAP_TLS_CACERTFILE: str | None = None

302 LDAP_ENABLE_STARTTLS: bool = False

303 LDAP_BASE_DN: str | None = None

304 LDAP_QUERY_BIND: str | None = None

305 LDAP_QUERY_PASSWORD: MaskedNoneString = None

306 LDAP_USER_FILTER: str | None = None

307 LDAP_ADMIN_FILTER: str | None = None

308 LDAP_ID_ATTRIBUTE: str = "uid"

309 LDAP_MAIL_ATTRIBUTE: str = "mail"

310 LDAP_NAME_ATTRIBUTE: str = "name"

311

312 @property

313 def LDAP_FEATURE(self) -> FeatureDetails:

314 description = None if self.LDAP_AUTH_ENABLED else "LDAP_AUTH_ENABLED is false"

315 required = {

316 "LDAP_SERVER_URL": self.LDAP_SERVER_URL,

317 "LDAP_BASE_DN": self.LDAP_BASE_DN,

318 "LDAP_ID_ATTRIBUTE": self.LDAP_ID_ATTRIBUTE,

319 "LDAP_MAIL_ATTRIBUTE": self.LDAP_MAIL_ATTRIBUTE,

320 "LDAP_NAME_ATTRIBUTE": self.LDAP_NAME_ATTRIBUTE,

321 }

322 not_none = None not in required.values()

323 if not not_none and not description: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true

324 missing_values = [key for (key, value) in required.items() if value is None]

325 description = f"Missing required values for {missing_values}"

326

327 return FeatureDetails(

328 enabled=self.LDAP_AUTH_ENABLED and not_none,

329 description=description,

330 )

331

332 @property

333 def LDAP_ENABLED(self) -> bool:

334 """Validates LDAP settings are all set"""

335 return self.LDAP_FEATURE.enabled

336

337 # ===============================================

338 # OIDC Configuration

339 OIDC_AUTH_ENABLED: bool = False

340 OIDC_CLIENT_ID: str | None = None

341 OIDC_CLIENT_SECRET: MaskedNoneString = None

342 OIDC_CONFIGURATION_URL: str | None = None

343 OIDC_SIGNUP_ENABLED: bool = True

344 OIDC_USER_GROUP: str | None = None

345 OIDC_ADMIN_GROUP: str | None = None

346 OIDC_AUTO_REDIRECT: bool = False

347 OIDC_PROVIDER_NAME: str = "OAuth"

348 OIDC_REMEMBER_ME: bool = False

349 OIDC_USER_CLAIM: str = "email"

350 OIDC_NAME_CLAIM: str = "name"

351 OIDC_GROUPS_CLAIM: str | None = "groups"

352 OIDC_SCOPES_OVERRIDE: str | None = None

353 OIDC_TLS_CACERTFILE: str | None = None

354

355 @property

356 def OIDC_REQUIRES_GROUP_CLAIM(self) -> bool:

357 return self.OIDC_USER_GROUP is not None or self.OIDC_ADMIN_GROUP is not None

358

359 @property

360 def OIDC_FEATURE(self) -> FeatureDetails:

361 description = None if self.OIDC_AUTH_ENABLED else "OIDC_AUTH_ENABLED is false"

362 required = {

363 "OIDC_CLIENT_ID": self.OIDC_CLIENT_ID,

364 "OIDC_CLIENT_SECRET": self.OIDC_CLIENT_SECRET,

365 "OIDC_CONFIGURATION_URL": self.OIDC_CONFIGURATION_URL,

366 "OIDC_USER_CLAIM": self.OIDC_USER_CLAIM,

367 }

368 not_none = None not in required.values()

369 if not not_none and not description: 369 ↛ 370line 369 didn't jump to line 370 because the condition on line 369 was never true

370 missing_values = [key for (key, value) in required.items() if value is None]

371 description = f"Missing required values for {missing_values}"

372

373 valid_group_claim = True

374 if self.OIDC_REQUIRES_GROUP_CLAIM and self.OIDC_GROUPS_CLAIM is None: 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true

375 if not description:

376 description = "OIDC_GROUPS_CLAIM is required when OIDC_USER_GROUP or OIDC_ADMIN_GROUP are provided"

377 valid_group_claim = False

378

379 return FeatureDetails(

380 enabled=self.OIDC_AUTH_ENABLED and not_none and valid_group_claim,

381 description=description,

382 )

383

384 @property

385 def OIDC_READY(self) -> bool:

386 """Validates OIDC settings are all set"""

387 return self.OIDC_FEATURE.enabled

388

389 # ===============================================

390 # OpenAI Configuration

391

392 OPENAI_BASE_URL: str | None = None

393 """The base URL for the OpenAI API. Leave this unset for most usecases"""

394 OPENAI_API_KEY: MaskedNoneString = None

395 """Your OpenAI API key. Required to enable OpenAI features"""

396 OPENAI_MODEL: str = "gpt-4o"

397 """Which OpenAI model to send requests to. Leave this unset for most usecases"""

398 OPENAI_CUSTOM_HEADERS: dict[str, str] = {}

399 """Custom HTTP headers to send with each OpenAI request"""

400 OPENAI_CUSTOM_PARAMS: dict[str, Any] = {}

401 """Custom HTTP parameters to send with each OpenAI request"""

402 OPENAI_ENABLE_IMAGE_SERVICES: bool = True

403 """Whether to enable image-related features in OpenAI"""

404 OPENAI_WORKERS: int = 2

405 """

406 Number of OpenAI workers per request. Higher values may increase

407 processing speed, but will incur additional API costs

408 """

409 OPENAI_SEND_DATABASE_DATA: bool = True

410 """

411 Sending database data may increase accuracy in certain requests,

412 but will incur additional API costs

413 """

414 OPENAI_REQUEST_TIMEOUT: int = 300

415 """

416 The number of seconds to wait for an OpenAI request to complete before cancelling the request

417 """

418

419 @property

420 def OPENAI_FEATURE(self) -> FeatureDetails:

421 description = None

422 if not self.OPENAI_API_KEY: 422 ↛ 424line 422 didn't jump to line 424 because the condition on line 422 was always true

423 description = "OPENAI_API_KEY is not set"

424 elif not self.OPENAI_MODEL:

425 description = "OPENAI_MODEL is not set"

426

427 return FeatureDetails(

428 enabled=bool(self.OPENAI_API_KEY and self.OPENAI_MODEL),

429 description=description,

430 )

431

432 @property

433 def OPENAI_ENABLED(self) -> bool:

434 """Validates OpenAI settings are all set"""

435 return self.OPENAI_FEATURE.enabled

436

437 # ===============================================

438 # Web Concurrency

439

440 WORKER_PER_CORE: int = 1

441 """Old gunicorn env for workers per core."""

442

443 UVICORN_WORKERS: int = 1

444 """Number of Uvicorn workers to run."""

445

446 @property

447 def WORKERS(self) -> int:

448 return max(1, self.WORKER_PER_CORE * self.UVICORN_WORKERS)

449

450 model_config = SettingsConfigDict(arbitrary_types_allowed=True, extra="allow", env_nested_delimiter="__")

451

452 # ===============================================

453 # TLS

454

455 TLS_CERTIFICATE_PATH: str | os.PathLike[str] | None = None

456 """Path where the certificate resides."""

457

458 TLS_PRIVATE_KEY_PATH: str | os.PathLike[str] | None = None

459 """Path where the private key resides."""

460

461

462def app_settings_constructor(data_dir: Path, production: bool, env_file: Path, env_encoding="utf-8") -> AppSettings:

463 """

464 app_settings_constructor is a factory function that returns an AppSettings object. It is used to inject the

465 required dependencies into the AppSettings object and nested child objects. AppSettings should not be instantiated

466 directly, but rather through this factory function.

467 """

468 secret_settings = {

469 "SECRET": determine_secrets(data_dir, ".secret", production),

470 "SESSION_SECRET": determine_secrets(data_dir, ".session_secret", production),

471 }

472 app_settings = AppSettings(

473 _env_file=env_file, # type: ignore

474 _env_file_encoding=env_encoding, # type: ignore

475 # `get_secrets_dir` must be called here rather than within `AppSettings`

476 # to avoid a circular import.

477 _secrets_dir=get_secrets_dir(), # type: ignore

478 **secret_settings,

479 )

480

481 app_settings.DB_PROVIDER = db_provider_factory(

482 app_settings.DB_ENGINE or "sqlite",

483 data_dir,

484 env_file=env_file,

485 env_encoding=env_encoding,

486 )

487

488 return app_settings