File Content: base.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import contextlib from typing import Any, TypedDict, Callable from threading import Thread, Event import sqlalchemy as sa from lvestats.orm import BurstingEventType, bursting_events_table from ..common import LveId, Timestamp def init_db_schema(engine: sa.engine.Engine) -> None: bursting_events_table.create(engine, checkfirst=True) def drop_db_schema(engine: sa.engine.Engine) -> None: bursting_events_table.drop(engine, checkfirst=True) class InBurstingEventRow(TypedDict): lve_id: LveId timestamp: Timestamp event_type: BurstingEventType @contextlib.contextmanager def thread_running(name: str, main: Callable[[Event], Any]): terminate = Event() (saver_thread := Thread( name=name, target=main, # NOTE(vlebedev): For details regarding how this interacts with termination using "atexit" callback, # have a look at https://stackoverflow.com/a/58928213/3344105 daemon=True, args=(terminate,), )).start() try: yield finally: terminate.set() saver_thread.join() Submit