Coverage for src/pydal2sql_core/types.py: 93%

1""" 

2Contains types for core.py. 

3""" 

4 

5import typing 

6import warnings 

7from typing import Any 

8 

9import pydal 

10from pydal.adapters import SQLAdapter as _SQLAdapter 

11from witchery import Empty 

12 

13SUPPORTED_DATABASE_TYPES = typing.Literal["psycopg2", "sqlite3", "pymysql"] 

14DATABASE_ALIASES_PSQL = typing.Literal["postgresql", "postgres", "psql"] 

15DATABASE_ALIASES_SQLITE = typing.Literal["sqlite"] 

16DATABASE_ALIASES_MYSQL = typing.Literal["mysql"] 

17 

18DATABASE_ALIASES = DATABASE_ALIASES_PSQL | DATABASE_ALIASES_SQLITE | DATABASE_ALIASES_MYSQL 

19SUPPORTED_DATABASE_TYPES_WITH_ALIASES = SUPPORTED_DATABASE_TYPES | DATABASE_ALIASES 

20 

21_SUPPORTED_OUTPUT_FORMATS = typing.Literal["default", "edwh-migrate"] 

22SUPPORTED_OUTPUT_FORMATS = _SUPPORTED_OUTPUT_FORMATS | None 

23DEFAULT_OUTPUT_FORMAT: SUPPORTED_OUTPUT_FORMATS = "default" 

24 

25 

26class SQLAdapter(_SQLAdapter): # type: ignore 

27 """ 

28 Typing friendly version of pydal's SQL Adapter. 

29 """ 

30 

31 

32empty = Empty() 

33 

34 

35class UniversalSet(dict): 

36 def __contains__(self, _) -> bool: 

37 return True 

38 

39 def __getitem__(self, item): 

40 try: 

41 return super().__getitem__(item) 

42 except KeyError: 

43 # e.g. for mapping `timestamp` as special field type 

44 return item 

45 

46 

47class CustomAdapter(SQLAdapter): 

48 """ 

49 Adapter that prevents actual queries. 

50 """ 

51 

52 drivers = ("sqlite3",) 

53 

54 def _log_attempt(self) -> None: 

55 from .state import state 

56 

57 if state.verbosity > 2: 

58 warnings.warn("Prevented attempt to execute query while migrating.") 

59 

60 @property 

61 def types(self): 

62 # special type that ensures 'x in types' is always true 

63 return UniversalSet(super().types) 

64 

65 def id_query(self, _: Any) -> Empty: # pragma: no cover 

66 """ 

67 Normally generates table._id != None. 

68 """ 

69 self._log_attempt() 

70 return empty 

71 

72 def execute(self, *_: Any, **__: Any) -> Empty: 

73 """ 

74 Normally executes an SQL query on the adapter. 

75 """ 

76 self._log_attempt() 

77 return empty 

78 

79 @property 

80 def cursor(self) -> Empty: 

81 """ 

82 Trying to connect to the database. 

83 """ 

84 self._log_attempt() 

85 return empty 

86 

87 

88class DummyDAL(pydal.DAL): # type: ignore 

89 """ 

90 Subclass of DAL that disables committing. 

91 """ 

92 

93 def commit(self) -> None: 

94 """ 

95 Do Nothing. 

96 """ 

97 

98 def __getattribute__(self, item: str) -> Any: 

99 """ 

100 Replace dal._adapter with a custom adapter that doesn't run queries. 

101 """ 

102 if item == "_adapter": 

103 return CustomAdapter(self, "", adapter_args={"driver": "sqlite3"}, driver_args="") 

104 

105 return super().__getattribute__(item) 

106 

107 def __call__(self, *_: Any, **__: Any) -> Empty: 

108 """ 

109 Prevents calling db() and thus creating a query. 

110 """ 

111 return empty 

112 

113 

114try: 

115 import typedal 

116 

117 class DummyTypeDAL(typedal.TypeDAL, DummyDAL): 

118 """ 

119 Variant of DummyDAL for TypeDAL. 

120 """ 

121 

122 def __init__(self, *args: Any, **settings: Any) -> None: 

123 """ 

124 Force TypeDAL to ignore project/env settings. 

125 """ 

126 # dummy typedal should not look at these settings: 

127 settings["use_pyproject"] = False 

128 settings["use_env"] = False 

129 if not settings.get("folder"): 

130 settings["folder"] = "/tmp/typedal2sql" 

131 

132 super().__init__(*args, **settings) 

133 

134except ImportError: # pragma: no cover 

135 DummyTypeDAL = DummyDAL # type: ignore