mission-peace/interview · error · KeyError

Key already exists

Error message

Key already exists

What it means

add_task raises KeyError("Key already exists") when the task is already present in entry_finder. Tasks must be unique in this priority queue; to change an existing task's priority use change_task_priority instead of add_task.

Solutions

  1. Use change_task_priority(priority, task) when the task may already exist
  2. Call remove_task(task) before add_task to replace the entry
  3. Guard with 'if task not in pq.entry_finder: pq.add_task(...)'

Example fix

// before
pq.add_task(priority, task)
// after
if task in pq.entry_finder:
    pq.change_task_priority(priority, task)
else:
    pq.add_task(priority, task)
Defensive patterns

Strategy: try-catch

Validate before calling

if task not in pq.entry_finder:
    pq.add_task(priority, task)
else:
    pq.change_task_priority(priority, task)

Type guard

def can_add(pq, task):
    return task not in pq.entry_finder

Try / catch

try:
    pq.add_task(priority, task)
except KeyError:
    pq.change_task_priority(priority, task)

Prevention

When it happens

Trigger: Calling add_task(priority, task) twice with the same task without removing it in between.

Common situations: Re-enqueueing a retried job without remove_task, re-running a seeding function on an existing queue, or using add_task where change_task_priority was intended.

Understand the failure class

Background: "already exists" / EEXIST / FileAlreadyExistsException: what the 'file already exists' error means and how to fix it — this error's family across 37 libraries.

Related errors

AI-assisted analysis of mission-peace/interview@94be5deb0c (2026-09-08). Data as JSON: /api/errors/f64a9d414c8b5abc. Report an issue: GitHub.

Appendix: source

Thrown at python/graph/priorityqueue.py:30

        if(is_min_heap is True):
            self.mul = 1
        else :
            self.mul = -1
         
    def contains_task(self, task):
        if task in self.entry_finder:
            return True
        else:
            return False

    def get_task_priority(self, task):
        if task in self.entry_finder:
            return (self.entry_finder[task])[0]
        raise ValueError("task does not exist")
        
    def add_task(self, priority, task):
        if task in self.entry_finder:
            raise KeyError("Key already exists")
        entry = [self.mul*priority, False, task]
        self.entry_finder[task] = entry
        heappush(self.pq, entry)

    def change_task_priority(self, priority, task):
        if task not in self.entry_finder:
            raise KeyError("Task not found")
        self.remove_task(task)
        entry = [self.mul*priority, False, task]
        self.entry_finder[task] = entry
        heappush(self.pq, entry)
 
    def remove_task(self, task):
        entry = self.entry_finder.pop(task)
        entry[1] = True

    def pop_task(self):
        while self.pq:

View on GitHub (pinned to 94be5deb0c)