Repeating the module name tends to be tedious and in many cases doesn't add much. For imports, I generally prefer from unittest import TestCase.Rename get to pop (you even call it this in the comment) for parity with sets (and python's heap implementation). ![]() Rename size to _len_ so you can do len(queue).lists and other builtins are falsy when empty Instead of empty define a _bool_ which returns True if the queue has items.I'd do assert removed_count in (0, 1) (tuples are preferred when you don't need to modify the object).If you want to return a bool, then name it discard (also for parity). If you name it remove for parity with how the builtin set does it, I'd have it raise KeyError if item isn't in the queue instead of returning a bool.I'm not sure if I would call it contains, but ntains(item, priority=5) reads better to me. contains is usually spelled _contains_.Thread = threading.Thread(target=add_to_queue, args=(), kwargs=) Queue: OrderedSetPriorityQueue = OrderedSetPriorityQueue()ĭef test_increase_priority(self) -> None:ĭef test_decrease_priority(self) -> None: Verify integrity of custom queue implementation """Internal function used to clean up unused data structures."""Īssert sorted(()) = sorted(self.elem_sets.keys())Ĭlass OrderedSetPriorityQueueTest(unittest.TestCase): """Pop the oldest item from the highest priority.""" Return sum(map(len, self.elem_sets.values())) """Return True if the queue is completely empty."""Īssert bool(self.elem_sets) = bool(ques) The entry with the old priority will be removed, If it already was queued but with a different priority, If is has not been queued yet, it is added and True is returned. """Returns False if item already is queued with the same priority. """Remove an item from the queue, disregarding its stored priority."""įor set_prio, elem_set in self.elem_ems():ĭef put(self, item: T, priority: int = 0) -> bool: """Check if the item is already queued with this exact priority.""" ![]() """Check if the item is already queued."""ĭef contains_with_prio(self, item: T, priority: int) -> bool: ndition_var = threading.Condition(threading.RLock()) ![]() """A thread-safe priority queue keeping its elements unique""" """A thread-safe priority queue keeping its elements uniqueįrom typing import DefaultDict, Deque, Set, Generic, TypeVarĬlass OrderedSetPriorityQueue(Generic): I would be happy to hear your ideas on how the code can be improved.
0 Comments
Leave a Reply. |