(self, config: StorageConfig)
| 18 | """Thin sync wrapper around async SQLExperienceStorage for Streamlit.""" |
| 19 | |
| 20 | def __init__(self, config: StorageConfig) -> None: |
| 21 | self._loop = asyncio.new_event_loop() |
| 22 | self._loop_ready = threading.Event() |
| 23 | self._closed = False |
| 24 | self._thread = threading.Thread(target=self._run_loop, daemon=True) |
| 25 | self._thread.start() |
| 26 | self._loop_ready.wait() |
| 27 | self._async = SQLExperienceStorage(config) |
| 28 | self._run(self._async.prepare()) |
| 29 | |
| 30 | def _run_loop(self) -> None: |
| 31 | asyncio.set_event_loop(self._loop) |
nothing calls this directly
no test coverage detected