(self, file_path: str, **kwargs: Any)
| 65 | ) |
| 66 | |
| 67 | def __init__(self, file_path: str, **kwargs: Any) -> None: |
| 68 | super().__init__(**kwargs) |
| 69 | self.filename = file_path |
| 70 | |
| 71 | self.pragmas = kwargs.copy() |
| 72 | self.pragmas.pop("connection_name", None) |
| 73 | self.pragmas.pop("fetch_inserted", None) |
| 74 | self.pragmas.setdefault("journal_mode", "WAL") |
| 75 | self.pragmas.setdefault("journal_size_limit", 16384) |
| 76 | self.pragmas.setdefault("foreign_keys", "ON") |
| 77 | |
| 78 | self._connection: aiosqlite.Connection | None = None |
| 79 | self._lock = asyncio.Lock() |
| 80 | |
| 81 | async def create_connection(self, with_db: bool) -> None: |
| 82 | if not self._connection: # pragma: no branch |
nothing calls this directly
no outgoing calls
no test coverage detected