NaturalHistoryMuseum / splitgill / #72

Committed 02 Aug 2024 09:12PM UTC coverage: 35.289% (-60.8%) from 96.047%

Build # #72

Build Type

push

coveralls-python

Committed by

jrdh

Commit Message

build: swap docker-compose for docker compose

Run Details

379 of 1074 relevant lines covered (35.29%)

0.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

43.24

/splitgill/locking.py

import platform
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Optional

from pymongo.collection import Collection
from pymongo.errors import DuplicateKeyError


class AlreadyLocked(Exception):
    """
    Exception that will be thrown when a lock is already acquired.
    """

    def __init__(self, lock_id: str):
        super().__init__(f"Lock '{lock_id}' is already locked")
        self.lock_id = lock_id


class LockManager:
    """
    Class for managing locks.
    """

    def __init__(self, lock_collection: Collection):
        """
        :param lock_collection: the collection to use for lock data
        """
        self.lock_collection = lock_collection
        # does nothing if this already exists
        self.lock_collection.create_index("lock_id", unique=True)

    def acquire(self, lock_id: str, raise_on_fail: bool = False, **kwargs) -> bool:
        """
        Acquire the lock with the given lock_id. If the lock can't be acquired, False is
        returned, if it can True is returned. If raise_on_fail is set to True, a
        AlreadyLocked exception is raised if the lock can't be acquired.

        Any additional keyword arguments provided are stored in the lock collection with
        the core lock metadata.

        :param lock_id: the ID of the lock to acquire
        :param raise_on_fail: if True, raises an AlreadyLocked exception if the lock
                              can't be acquired. Default: False.
        :return: True if the lock was acquired, False if not
        """
        try:
            doc = {
                "lock_id": lock_id,
                "locked_at": datetime.now(timezone.utc),
                "locked_by": platform.node(),
            }
            if kwargs:
                doc["data"] = kwargs
            self.lock_collection.insert_one(doc)
        except DuplicateKeyError:
            if raise_on_fail:
                raise AlreadyLocked(lock_id)
            return False
        return True

    def release(self, lock_id: str):
        """
        Release the lock with the given lock_id. If the lock_id isn't locked, does
        nothing.

        :param lock_id: the ID of the lock to release
        """
        self.lock_collection.delete_one({"lock_id": lock_id})

    def is_locked(self, lock_id: str) -> bool:
        """
        Check if the given lock_id is locked or not.

        :param lock_id: the ID of the lock to check
        :return: True if the lock is currently acquired, False if not
        """
        return self.get_metadata(lock_id) is not None

    def get_metadata(self, lock_id: str) -> Optional[dict]:
        """
        Returns the doc stored in the lock collection for the given lock ID, if there is
        one.

        :param lock_id:
        :return:
        """
        return self.lock_collection.find_one({"lock_id": lock_id})

    @contextmanager
    def lock(self, lock_id: str, **kwargs):
        """
        Context manager to safely acquire and release a lock with the given lock ID. If
        the lock is already acquired, raises an AlreadyLocked exception.

        Any additional keyword arguments provided are stored in the lock collection with
        the core lock metadata.

        :param lock_id: ID of the lock to acquire and release
        """
        self.acquire(lock_id, raise_on_fail=True, **kwargs)
        try:
            yield
        finally:
            self.release(lock_id)

1	import platform	1✔
2	from contextlib import contextmanager	1✔
3	from datetime import datetime, timezone	1✔
4	from typing import Optional	1✔
5
6	from pymongo.collection import Collection	1✔
7	from pymongo.errors import DuplicateKeyError	1✔
8
9
10	class AlreadyLocked(Exception):	1✔
11	"""
12	Exception that will be thrown when a lock is already acquired.
13	"""
14
15	def __init__(self, lock_id: str):	1✔
16	super().__init__(f"Lock '{lock_id}' is already locked")	×
17	self.lock_id = lock_id	×
18
19
20	class LockManager:	1✔
21	"""
22	Class for managing locks.
23	"""
24
25	def __init__(self, lock_collection: Collection):	1✔
26	"""
27	:param lock_collection: the collection to use for lock data
28	"""
29	self.lock_collection = lock_collection	×
30	# does nothing if this already exists
31	self.lock_collection.create_index("lock_id", unique=True)	×
32
33	def acquire(self, lock_id: str, raise_on_fail: bool = False, **kwargs) -> bool:	1✔
34	"""
35	Acquire the lock with the given lock_id. If the lock can't be acquired, False is
36	returned, if it can True is returned. If raise_on_fail is set to True, a
37	AlreadyLocked exception is raised if the lock can't be acquired.
38
39	Any additional keyword arguments provided are stored in the lock collection with
40	the core lock metadata.
41
42	:param lock_id: the ID of the lock to acquire
43	:param raise_on_fail: if True, raises an AlreadyLocked exception if the lock
44	can't be acquired. Default: False.
45	:return: True if the lock was acquired, False if not
46	"""
47	try:	×
48	doc = {	×
49	"lock_id": lock_id,
50	"locked_at": datetime.now(timezone.utc),
51	"locked_by": platform.node(),
52	}
53	if kwargs:	×
54	doc["data"] = kwargs	×
55	self.lock_collection.insert_one(doc)	×
56	except DuplicateKeyError:	×
57	if raise_on_fail:	×
58	raise AlreadyLocked(lock_id)	×
59	return False	×
60	return True	×
61
62	def release(self, lock_id: str):	1✔
63	"""
64	Release the lock with the given lock_id. If the lock_id isn't locked, does
65	nothing.
66
67	:param lock_id: the ID of the lock to release
68	"""
69	self.lock_collection.delete_one({"lock_id": lock_id})	×
70
71	def is_locked(self, lock_id: str) -> bool:	1✔
72	"""
73	Check if the given lock_id is locked or not.
74
75	:param lock_id: the ID of the lock to check
76	:return: True if the lock is currently acquired, False if not
77	"""
78	return self.get_metadata(lock_id) is not None	×
79
80	def get_metadata(self, lock_id: str) -> Optional[dict]:	1✔
81	"""
82	Returns the doc stored in the lock collection for the given lock ID, if there is
83	one.
84
85	:param lock_id:
86	:return:
87	"""
88	return self.lock_collection.find_one({"lock_id": lock_id})	×
89
90	@contextmanager	1✔
91	def lock(self, lock_id: str, **kwargs):	1✔
92	"""
93	Context manager to safely acquire and release a lock with the given lock ID. If
94	the lock is already acquired, raises an AlreadyLocked exception.
95
96	Any additional keyword arguments provided are stored in the lock collection with
97	the core lock metadata.
98
99	:param lock_id: ID of the lock to acquire and release
100	"""
101	self.acquire(lock_id, raise_on_fail=True, **kwargs)	×
102	try:	×
103	yield	×
104	finally:
105	self.release(lock_id)	×

NaturalHistoryMuseum / splitgill / #72

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous