haf
03/11/2021, 3:58 PMUnexpected 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)
#!/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(),
)
haf
03/11/2021, 3:58 PMhaf
03/11/2021, 4:23 PMhaf
03/11/2021, 4:30 PM