(self, with_db: bool)
| 79 | self._lock = asyncio.Lock() |
| 80 | |
| 81 | async def create_connection(self, with_db: bool) -> None: |
| 82 | if not self._connection: # pragma: no branch |
| 83 | self._connection = await aiosqlite.connect(self.filename, isolation_level=None) |
| 84 | self._connection._conn.row_factory = sqlite3.Row |
| 85 | for pragma, val in self.pragmas.items(): |
| 86 | cursor = await self._connection.execute(f"PRAGMA {pragma}={val}") |
| 87 | await cursor.close() |
| 88 | await self._post_connect() |
| 89 | self.log.debug( |
| 90 | "Created connection %s with params: filename=%s %s", |
| 91 | self._connection, |
| 92 | self.filename, |
| 93 | " ".join(f"{k}={v}" for k, v in self.pragmas.items()), |
| 94 | ) |
| 95 | |
| 96 | async def close(self) -> None: |
| 97 | if self._connection: |
no test coverage detected