(self)
| 114 | self.connect() |
| 115 | |
| 116 | def connect(self): |
| 117 | self.connection = sqlite3.connect(self.database, timeout=10.0) |
| 118 | self.connection.enable_load_extension(True) |
| 119 | self.connection.execute("PRAGMA foreign_keys = ON;") |
| 120 | self.connection.execute("PRAGMA journal_mode=WAL;") |
| 121 | self.connection.execute("PRAGMA auto_vacuum=full;") |
| 122 | self.cursor = self.connection.cursor() |
| 123 | |
| 124 | if self.zstd_enabled: |
| 125 | if sqlite_zstd is None: |
| 126 | raise ValueError("sqlite_zstd library not found.") |
| 127 | |
| 128 | sqlite_zstd.load(self.connection) |
| 129 | self.enable_zstd() |
| 130 | |
| 131 | def ensure_connection(self): |
| 132 | if self.connection is None or self.cursor is None: |
no test coverage detected