I’m running into an error using

flow.replace

, which seems to have to do with the fact that

flow.remove(task)

doesn’t remove

task

from `flow.slugs`:

from prefect import Parameter, Flow, Task

p1 = Parameter('p')
p2 = Parameter('p')
t1 = Task()
t2 = Task()

flow = Flow('flow')
flow.add_task(p1)
flow.add_task(t1)

print(flow.slugs)
# {<Parameter: p>: 'p', <Task: Task>: 'Task-1'}

flow.replace(t1, t2)
print(flow.slugs)
# {<Parameter: p>: 'p', <Task: Task>: 'Task-1', <Task: Task>: 'Task-1'}

flow.replace(p1, p2)
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-7-5ec138031eb6> in <module>
----> 1 flow.replace(p1, p2)

/opt/conda/envs/drugdiscovery/lib/python3.6/site-packages/prefect/core/flow.py in replace(self, old, new, validate)
    291         # update tasks
    292         self.tasks.remove(old)
--> 293         self.add_task(new)
    294 
    295         self._cache.clear()

/opt/conda/envs/drugdiscovery/lib/python3.6/site-packages/prefect/core/flow.py in add_task(self, task)
    482                 raise ValueError(
    483                     'A task with the slug "{}" already exists in this '
--> 484                     "flow.".format(task.slug)
    485                 )
    486             self.slugs[task] = task.slug or self._generate_task_slug(task)

ValueError: A task with the slug "p" already exists in this flow.

Hey @Ben Fogelson - this appears to be a bug! I’ll open an issue on GitHub for you to track

Thanks @Chris White! FYI did some testing, the bug appears to have been introduced between 0.12.0 and 0.12.1