Has anyone encountered and solved `Unexpected erro...
# ask-community
h
Has anyone encountered and solved
Unexpected error while running flow: KeyError('Task slug resolve_profiles_dir-1 not found in the current Flow; this is usually caused by changing the Flow without reregistering it with the Prefect API.')
— I set up all agents/cli:s two days ago at their latest versions, and the agent is a k8s one. I'm using Prefect cloud. This happens when I register the flow as such and then run it from the UI (I verified the UI archives the old version and I'm triggering a run of the updated version)
Copy code
#!/usr/bin/env python

from logging import getLogger
from datetime import timedelta
from os import getenv
from pathlib import Path
from pendulum import today
from prefect.engine.state import Failed
from prefect.schedules import IntervalSchedule
from prefect.storage import Docker
from prefect.utilities.notifications import slack_notifier
from prefect.utilities.storage import extract_flow_from_file

logger = getLogger("dbt.deploy")

with open('requirements.txt') as file:
    packages = list(line.strip() for line in file.readlines())

docker = Docker(
    registry_url="europe-docker.pkg.dev/projecthere/cd",
    # dockerfile='Dockerfile', # Uncomment to use own Dockerfile with e.g. dependencies installed
    image_name="analytics-dbt",
    image_tag="0.14.11",
    python_dependencies=packages
)

slack = slack_notifier(
    only_states=[Failed],
    webhook_secret='SLACK_WEBHOOK_URL')

every_hour = IntervalSchedule(
    start_date=today('utc'),
    interval=timedelta(hours=1))

flows = sorted(Path('flows').glob('*.py'))

# Add flows
for file in flows:
    flow = extract_flow_from_file(file_path=file)
    <http://logger.info|logger.info>('Extracted flow from file before build')

    docker.add_flow(flow)

# Build storage with all flows
docker = docker.build()

# Update storage in flows and register
for file in flows:
    flow = extract_flow_from_file(file_path=file)
    <http://logger.info|logger.info>('Extracted flow from file after build')

    flow.storage = docker
    flow.state_handlers.append(slack)
    flow.schedule = every_hour

    <http://logger.info|logger.info>('Registering...')
    flow.register(
        project_name='dbt',
        build=False,
        labels=['prod'],
        idempotency_key=flow.serialized_hash(),
    )
This thread is related: https://prefect-community.slack.com/archives/CL09KU1K7/p1606764338459500 but doesn't provide a clear answer to what it might be.
I have a lead; this version doesn't talk about the base image's version, but the tag this docker image gets?
Yes indeed it is. This was the problem, because it caused the cached / already downloaded image to be run instead of the newly built (retagged) image.