NaturalHistoryMuseum / splitgill / #76

Committed 04 Aug 2024 07:46PM UTC coverage: 34.976% (-59.7%) from 94.693%

Build # #76

Build Type

push

coveralls-python

Committed by

jrdh

Commit Message

feat: remove default config values

BREAKING CHANGE: remove default config values

Run Details

3 of 19 new or added lines in 3 files covered. (15.79%)

613 existing lines in 13 files now uncovered.

369 of 1055 relevant lines covered (34.98%)

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✔
UNCOV 16	super().__init__(f"Lock '{lock_id}' is already locked")	×
UNCOV 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	"""
UNCOV 29	self.lock_collection = lock_collection	×
30	# does nothing if this already exists
UNCOV 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	"""
UNCOV 47	try:	×
UNCOV 48	doc = {	×
49	"lock_id": lock_id,
50	"locked_at": datetime.now(timezone.utc),
51	"locked_by": platform.node(),
52	}
UNCOV 53	if kwargs:	×
UNCOV 54	doc["data"] = kwargs	×
UNCOV 55	self.lock_collection.insert_one(doc)	×
UNCOV 56	except DuplicateKeyError:	×
UNCOV 57	if raise_on_fail:	×
UNCOV 58	raise AlreadyLocked(lock_id)	×
UNCOV 59	return False	×
UNCOV 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	"""
UNCOV 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	"""
UNCOV 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	"""
UNCOV 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	"""
UNCOV 101	self.acquire(lock_id, raise_on_fail=True, **kwargs)	×
UNCOV 102	try:	×
UNCOV 103	yield	×
104	finally:
UNCOV 105	self.release(lock_id)	×

NaturalHistoryMuseum / splitgill / #76

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