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

1from abc import ABC, abstractmethod 

2from pathlib import Path 

3from urllib import parse as urlparse 

4 

5from pydantic import BaseModel, PostgresDsn 

6from pydantic_settings import BaseSettings, SettingsConfigDict 

7 

8 

9class AbstractDBProvider(ABC): 

10 @property 

11 @abstractmethod 

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

13 

14 @property 

15 @abstractmethod 

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

17 

18 

19class SQLiteProvider(AbstractDBProvider, BaseModel): 

20 data_dir: Path 

21 prefix: str = "" 

22 

23 @property 

24 def db_path(self): 

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

26 

27 @property 

28 def db_url(self) -> str: 

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

30 

31 @property 

32 def db_url_public(self) -> str: 

33 return self.db_url 

34 

35 

36class PostgresProvider(AbstractDBProvider, BaseSettings): 

37 POSTGRES_USER: str = "mealie" 

38 POSTGRES_PASSWORD: str = "mealie" 

39 POSTGRES_SERVER: str = "postgres" 

40 POSTGRES_PORT: str = "5432" 

41 POSTGRES_DB: str = "mealie" 

42 POSTGRES_URL_OVERRIDE: str | None = None 

43 

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

45 

46 @property 

47 def db_url(self) -> str: 

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 

74 def db_url_public(self) -> str: 

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: 

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

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

92 else: 

93 return SQLiteProvider(data_dir=data_dir)