Skip to content
Snippets Groups Projects

msobczyk/handle signals in docker entrypoint

Merged Marcin Sobczyk requested to merge msobczyk/handle-signals-in-docker-entrypoint into develop
Compare and Show latest version
3 files
+ 24
26
Compare changes
  • Side-by-side
  • Inline
Files
3
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING
from clive_local_tools.waiters import wait_for
if TYPE_CHECKING:
from clive.__private.core.beekeeper import Beekeeper
async def wait_for_beekeeper_to_close(beekeeper: Beekeeper, timeout: float = 1.0) -> None:
async def __wait_for_beekeeper_to_close() -> None:
while beekeeper.is_already_running_locally():
await asyncio.sleep(0.1)
try:
await asyncio.wait_for(__wait_for_beekeeper_to_close(), timeout=timeout)
except asyncio.TimeoutError:
raise AssertionError(f"Beekeeper was not closed in expected time of {timeout:.2f}") from None
await wait_for(lambda: not beekeeper.is_already_running_locally(), "Beekeeper was not closed", timeout=timeout)
Loading