Coverage for src/pytest_samples/database/_engine.py: 100%

1import logging as _logging 

2import os as _os 

3import os.path as _ospath 

4import sqlalchemy as _sqlalchemy 

5 

6from sqlalchemy import Engine as _Engine 

7 

8from . import _defs 

9from . import _exceptions 

10from ._session import Session as _Session 

11 

12 

13_logger = _logging.getLogger(__name__) 

14"""The logger for this module.""" 

15 

16 

17class EngineBase: 

18 """Base class for the engine which wraps the sqlalchemy engine.""" 

19 

20 __slots__ = ("_engine", "_disposed") 

21 

22 def __init__(self, inner: _Engine) -> None: 

23 """Initialize the `_EngineBase`. 

24 

25 Args: 

26 inner (_Engine): The underlying `sqlalchemy` Engine. 

27 """ 

28 self._engine = inner 

29 """Whether the `setup_tables` was called to set up the 

30 tables. 

31 """ 

32 # It is important to track this since it seems like sqlalchemy 

33 # will just reopen the engine if any action is performed after 

34 # dispose is called. 

35 self._disposed: bool = False 

36 """Whther `dispose` was called.""" 

37 

38 def _ensure_not_disposed(self) -> None: 

39 """Raise an exception if the `Engine` instance was already 

40 disposed. 

41 

42 Raises: 

43 EngineDisposedError: If the instance was already disposed. 

44 """ 

45 if self._disposed: 

46 raise _exceptions.EngineDisposedError( 

47 "The Engine instance was already disposed." 

48 ) 

49 

50 # It does not make sense to implement this in a context manager 

51 # since it is not possible to use this in a context anyway. Instead, 

52 # the object will be torn down in a different hook function than 

53 # it was created. 

54 def dispose(self) -> None: 

55 """Close the engine.""" 

56 self._engine.dispose() 

57 self._disposed = True 

58 

59 def setup_tables(self) -> None: 

60 """Initialize the tables. 

61 

62 Raises: 

63 DatabaseError: If the database is corrupted. 

64 EngineDisposedError: If the instance was already disposed. 

65 """ 

66 self._ensure_not_disposed() 

67 _defs.create_tables(self._engine) 

68 

69 def new_session(self) -> _Session: 

70 """Start a new session. 

71 

72 Raises: 

73 EngineDisposedError: If the instance was already disposed. 

74 EngineNotInitializedError: If the `Engine` instance was not 

75 initialized. 

76 

77 Returns: 

78 Session: A new session object. 

79 """ 

80 self._ensure_not_disposed() 

81 return _Session(self._engine) 

82 

83 

84class Engine(EngineBase): 

85 """The main engine for database connections.""" 

86 

87 __slots__ = ("_path",) 

88 

89 def __init__(self, path: str) -> None: 

90 """Initialize a new `Engine` given the file path. 

91 

92 Args: 

93 path (str): The file path. 

94 

95 Raises: 

96 IsADirectoryError: If the provided path points to an 

97 existing directory. 

98 RelativePathError: If the provided path is a relative path. 

99 ValueError: If `path` is invalid, for example if it is the 

100 empty `str`. 

101 """ 

102 if len(path) == 0: 

103 raise ValueError("The database path was empty.") 

104 # This allows for the creation of a file named :memory:, which 

105 # would otherwise create an in-memory database. 

106 if not _ospath.isabs(path): 

107 raise _exceptions.RelativePathError( 

108 "The provided path is a relative path." 

109 ) 

110 if _ospath.isdir(path): 

111 raise IsADirectoryError( 

112 "The database path points to a directory." 

113 ) 

114 self._path = path 

115 engine = self.create_engine(path) 

116 super().__init__(engine) 

117 

118 @classmethod 

119 def create_engine(cls, path: str) -> _Engine: 

120 """Create the inner engine given the path. 

121 

122 Args: 

123 path (str): The path to the database file. 

124 

125 Returns: 

126 Engine: The sqlalchemy engine for the connections. 

127 """ 

128 return _sqlalchemy.create_engine(f"sqlite:///{path}") 

129 

130 def truncate_database_file(self) -> None: 

131 """Truncate the database file. 

132 

133 Raises: 

134 IsADirectoryError: If the provided path is a directory. 

135 FileNotFoundError: If the file does not exist. 

136 """ 

137 _logger.info("Truncating database file.") 

138 _os.truncate(self._path, 0)