util.py 864 Bytes
Newer Older
1
from logging import getLogger
segfault's avatar
segfault committed
2
import threading
3 4 5 6
import fcntl
from contextlib import contextmanager
from typing import TextIO

segfault's avatar
segfault committed
7 8 9
from gi.repository import GLib


10 11 12
logger = getLogger(__name__)


segfault's avatar
segfault committed
13 14 15 16 17 18 19 20
def run_threaded(function, *args):
    thread = threading.Thread(target=function, args=args)
    thread.start()


def process_mainloop_events():
    context = GLib.MainLoop().get_context()
    while context.pending():
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36
        context.iteration()


@contextmanager
def open_locked(path, *args, **kwargs) -> TextIO:
    with open(path, *args, **kwargs) as f:
        try:
            logger.log(5, "Acquiring file lock on %r", path)
            fcntl.flock(f, fcntl.LOCK_EX)
            logger.log(5, "Acquired file lock on %r", path)

            yield f

        finally:
            logger.log(5, "Releasing file lock on %r", path)
            fcntl.flock(f, fcntl.LOCK_UN)