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

54 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:32 +0000

1from abc import ABC, abstractmethod 1a

2from pathlib import Path 1a

3from urllib import parse as urlparse 1a

4 

5from pydantic import BaseModel, PostgresDsn 1a

6from pydantic_settings import BaseSettings, SettingsConfigDict 1a

7 

8 

9class AbstractDBProvider(ABC): 1a

10 @property 1a

11 @abstractmethod 1a

12 def db_url(self) -> str: ... 12 ↛ exitline 12 didn't return from function 'db_url' because 1a

13 

14 @property 1a

15 @abstractmethod 1a

16 def db_url_public(self) -> str: ... 16 ↛ exitline 16 didn't return from function 'db_url_public' because 1a

17 

18 

19class SQLiteProvider(AbstractDBProvider, BaseModel): 1a

20 data_dir: Path 1a

21 prefix: str = "" 1a

22 

23 @property 1a

24 def db_path(self): 1a

25 return self.data_dir / f"{self.prefix}mealie.db" 1a

26 

27 @property 1a

28 def db_url(self) -> str: 1a

29 return f"sqlite:///{self.db_path.absolute()!s}" 1a

30 

31 @property 1a

32 def db_url_public(self) -> str: 1a

33 return self.db_url 

34 

35 

36class PostgresProvider(AbstractDBProvider, BaseSettings): 1a

37 POSTGRES_USER: str = "mealie" 1a

38 POSTGRES_PASSWORD: str = "mealie" 1a

39 POSTGRES_SERVER: str = "postgres" 1a

40 POSTGRES_PORT: str = "5432" 1a

41 POSTGRES_DB: str = "mealie" 1a

42 POSTGRES_URL_OVERRIDE: str | None = None 1a

43 

44 model_config = SettingsConfigDict(arbitrary_types_allowed=True, extra="allow") 1a

45 

46 @property 1a

47 def db_url(self) -> str: 1a

48 if self.POSTGRES_URL_OVERRIDE: 

49 url = self.POSTGRES_URL_OVERRIDE 

50 

51 scheme, remainder = url.split("://", 1) 

52 if scheme != "postgresql": 

53 raise ValueError("POSTGRES_URL_OVERRIDE scheme must be postgresql") 

54 

55 remainder = remainder.split(":", 1)[1] 

56 password = remainder[: remainder.rfind("@")] 

57 quoted_password = urlparse.quote(password) 

58 

59 safe_url = url.replace(password, quoted_password) 

60 

61 return safe_url 

62 

63 return str( 

64 PostgresDsn.build( 

65 scheme="postgresql", 

66 username=self.POSTGRES_USER, 

67 password=self.POSTGRES_PASSWORD, 

68 host=f"{self.POSTGRES_SERVER}:{self.POSTGRES_PORT}", 

69 path=f"{self.POSTGRES_DB or ''}", 

70 ) 

71 ) 

72 

73 @property 1a

74 def db_url_public(self) -> str: 1a

75 if self.POSTGRES_URL_OVERRIDE: 

76 return "Postgres Url Overridden" 

77 

78 return str( 

79 PostgresDsn.build( 

80 scheme="postgresql", 

81 username="******", 

82 password="******", 

83 host=f"{self.POSTGRES_SERVER}:{self.POSTGRES_PORT}", 

84 path=f"{self.POSTGRES_DB or ''}", 

85 ) 

86 ) 

87 

88 

89def db_provider_factory(provider_name: str, data_dir: Path, env_file: Path, env_encoding="utf-8") -> AbstractDBProvider: 1a

90 if provider_name == "postgres": 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true1a

91 return PostgresProvider(_env_file=env_file, _env_file_encoding=env_encoding) 

92 else: 

93 return SQLiteProvider(data_dir=data_dir) 1a