Untitled

mail@pastecode.io avatar
unknown
plain_text
19 days ago
1.2 kB
2
Indexable
Never
from collections import deque
from in_memory_db import InMemoryDB

class InMemoryDBImpl(InMemoryDB):
    # ... (rest of the code is the same as before)

    def unlock(self, key: str, caller_id: str = None) -> str | None:
        if key not in self.locks:
            return "invalid_request"  # Key is not locked

        if caller_id is not None and self.locks[key] != caller_id:
            return None  # Caller doesn't own the lock

        released_user = self.locks.pop(key)
        if key not in self.lock_queues or not self.lock_queues[key]:
            # No one is waiting in the queue
            return "released"

        # Assign the lock to the next user in the queue
        next_user = self.lock_queues[key].popleft()
        self.locks[key] = next_user
        return "released"

    def set_or_inc(self, key: str, field: str, value: int, caller_id: str = None) -> int | None:
        if key in self.locks and (caller_id is None or self.locks[key] != caller_id):
            return None  # Ignore if locked by another user or no caller_id provided

        return self._set_or_inc_internal(key, field, value)

    # ... (rest of the code is the same as before)
Leave a Comment