(self)
| 1371 | |
| 1372 | @property |
| 1373 | def db(self) -> sqlite3.Connection: |
| 1374 | assert self._db_lock.locked() |
| 1375 | if self._db is not None: |
| 1376 | return self._db |
| 1377 | db_allowed = self.get_ceph_option("mgr_pool") |
| 1378 | if not db_allowed: |
| 1379 | raise MgrDBNotAllowed() |
| 1380 | self._db = self.open_db() |
| 1381 | if self._db is None: |
| 1382 | raise MgrDBNotReady() |
| 1383 | return self._db |
| 1384 | |
| 1385 | @contextmanager |
| 1386 | def exclusive_db_access(self) -> Iterator[sqlite3.Connection]: |
nothing calls this directly
no test coverage detected