@Marvin File "/home/olveh/envs/myproject/lib/python3.12/site-packages/prefect/client/base.py", line 163, in raise_for_status raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url 'http://127.0.0.1:4200/api/deployments/3f752b96-d8fc-497d-9a9d-1fa3074d40ef' Response: {'detail': 'Please provide a slug for each schedule in your request to ensure schedules are updated correctly.'} For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/422

✅ thought for 362.2 seconds

@Marvin so the deploy code i use have worked with a lot of other flows, but fails for this one. the schedules it tries to write is a list of length 1 [CronSchedule(cron='22 22 * * 1-5', timezone='Europe/Stockholm', day_or=True)]

✅ thought for 68.2 seconds

@Marvin Actually i have been able to override (and inherit) existing schedules for other deploys without slug names, and for this one i even tried to give it one in the ui after the issue first appeared. Here is my code: from typing import Callable import os import inspect import argparse from prefect import flow from prefect.client.orchestration import SyncPrefectClient from prefect.client.schemas.objects import DeploymentSchedule from prefect.exceptions import PrefectHTTPStatusError, ObjectNotFound from prefect.flows import Flow as FlowClass from prefect_github import GitHubCredentials from prefect.runner.storage import GitRepository from brrr.orchestration import prefect_constants as pfc class PrefectFlowDeployer(): def __init__( self, python_function: Callable, production: bool=False, work_pool_name: str=None, branch: str=None, image: str = None, optional_tags: list | str | None = None ): if production: work_pool_name = pfc.production_pool if not work_pool_name else work_pool_name image = pfc.docker_image_repository_prod if not image else image branch = pfc.git_prod_branch if not branch else branch else: work_pool_name = pfc.dev_pool if not work_pool_name else work_pool_name image = pfc.docker_image_repository_dev if not image else image branch = pfc.git_dev_branch if not branch else branch self.python_function = python_function self.work_pool_name = work_pool_name self.branch = branch self.image = image self.production = production self.schedules = None self.tags = self._create_deployment_tags(production, optional_tags) def _create_deployment_name(self): return self.input_flow.name + '_env_' + ('production' if self.production else 'development') + '_branch_' + self.branch @staticmethod def _normalize_tags(tags: list | str | None) -> list: """Normalize tags input to a list. Args: tags: Can be None, a string, or a list of strings Returns: A list of strings, or empty list if input is None """ if tags is None: return [] elif isinstance(tags, str): return [tags] elif isinstance(tags, list): return tags def _create_deployment_tags(self, production: bool, optional_tags: list | str | None = None) -> list: """Create deployment tags with mandatory environment tag and optional additional tags. Args: production: Whether this is a production deployment optional_tags: Optional additional tags (can be string or list) Returns: List of tags including mandatory environment tag and any optional tags """ # Mandatory environment tag env_tag = 'production' if production else 'development' # Normalize optional tags to list additional_tags = self._normalize_tags(optional_tags) # Combine tags (env tag first, then optional tags) return [env_tag] + additional_tags def set_deployment_schedule(self) -> None: if self.production: with SyncPrefectClient(api=pfc.prefect_api_url) as client: try: deployment = client.read_deployment_by_name(f"{self.input_flow.name}/{self.deployment_name}") deployment_schedules = client.read_deployment_schedules(deployment_id=deployment.id) # Extract the actual schedule objects from DeploymentSchedule objects self.schedules = [] for deployment_schedule in deployment_schedules: if hasattr(deployment_schedule, 'schedule') and deployment_schedule.schedule: self.schedules.append(deployment_schedule.schedule) else: # Fallback: try to access the schedule directly if it's stored differently for schedule_type in ['cron', 'interval', 'rrule']: schedule = getattr(deployment_schedule, schedule_type, None) if schedule: self.schedules.append(schedule) break except ObjectNotFound as exc: print("Deployment not found in production - no schedule inheritance") pass def deploy_flow_docker(self) -> None: """ Deploy the flow using the configured settings. """ if not hasattr(self, 'input_flow') or not isinstance(self.input_flow, FlowClass): raise ValueError('input_flow must exist and be a FlowClass object') self.input_flow.deploy( name=self.deployment_name, work_pool_name=self.work_pool_name, image=self.image, build=False, schedules=self.schedules, tags=self.tags, ) def flow_from_source_auto(self) -> None: """ Create a flow from source using Prefect 3.4 patterns with automatic entrypoint detection. This version automatically detects the calling file and uses it for the entrypoint. Args: input_function: Python function to get entrypoint from (optional) branch: Git branch (default: 'master'). Give eg 'development' to build from development branch Returns: Flow object ready for deployment """ # Get the calling frame to determine the entrypoint # Walk up the call stack until we find a file that's not this helper file current_frame = inspect.currentframe() calling_file = None # Walk up the call stack while current_frame: current_frame = current_frame.f_back if current_frame and '__file__' in current_frame.f_globals: potential_file = current_frame.f_globals['__file__'] # Skip this helper file itself if not potential_file.endswith('prefect_deploy_helpers.py'): calling_file = potential_file break if not calling_file: raise RuntimeError("Could not determine calling file for entrypoint") # Convert absolute path to relative path from current working directory cwd = os.getcwd() try: entrypoint = f"{os.path.relpath(calling_file, cwd)}:{self.python_function.name}" except ValueError: # If relative path fails, use absolute path entrypoint = f"{calling_file}:{self.python_function.name}" # Load the repository block using Prefect's block loading mechanism source = GitRepository( url=pfc.git_https_repo_url, credentials=GitHubCredentials.load(pfc.git_credentials_block_name), branch=self.branch ) self.input_flow = flow.from_source( source=source, entrypoint=entrypoint, ) def _parse_deployment_arguments(self, description: str = "Deploy Prefect flow") -> argparse.Namespace: """Parse command line arguments for deployment configuration.""" parser = argparse.ArgumentParser(description=description) # Deployment environment parser.add_argument('--production', action='store_true', help='Deploy to production environment (default: False)') # Work pool configuration parser.add_argument('--work-pool-name', type=str, default=None, help='Override work pool name (default: auto-selected based on production flag)') # Git configuration parser.add_argument('--branch', type=str, default=None, help='Git branch to deploy from (default: auto-selected based on production flag)') # Docker image configuration parser.add_argument('--image', type=str, default=None, help='Docker image to use (default: auto-selected based on production flag)') # Deployment tags parser.add_argument('--optional-tags', type=str, nargs='*', default=None, help='Optional deployment tags (space-separated). Env tag (prod/dev) is always added automatically.') return parser.parse_args() def deploy(self, name: str = None) -> None: """ Deploy the flow using the configured settings. Args: name: Optional custom name for the deployment """ self.flow_from_source_auto() self.deployment_name = name if name else self._create_deployment_name() self.set_deployment_schedule() self.deploy_flow_docker() @classmethod def from_args( cls, python_function: Callable, description: str = "Deploy Prefect flow", optional_tags: list | str | None = None ) -> 'PrefectFlowDeployer': """ Create a PrefectFlowDeployer instance from command-line arguments. Args: python_function: The flow function to deploy description: Description for the argument parser help text optional_tags: Optional tags to add to the deployment (in addition to CLI --optional-tags) Returns: PrefectFlowDeployer instance configured from command-line arguments """ # Create a temporary instance to use the parsing method temp_instance = cls(python_function=python_function) args = temp_instance._parse_deployment_arguments(description) # Combine CLI tags with programmatic tags combined_tags = cls._normalize_tags(optional_tags) if args.optional_tags: combined_tags.extend(args.optional_tags) return cls( python_function=python_function, production=args.production, work_pool_name=args.work_pool_name, branch=args.branch, image=args.image, optional_tags=combined_tags if combined_tags else None ) def deploy_flow_with_args( python_function: Callable, description: str = "Deploy Prefect flow", optional_tags: list | str | None = None ) -> None: """ Deploy a Prefect flow with command-line argument parsing. This function handles both cases - with or without command-line arguments. Args: python_function: The flow function to deploy description: Description for the argument parser help text optional_tags: Optional deployment tags (can be string or list). These will be combined with any --optional-tags provided via CLI. Command-line arguments: --production: Deploy to production environment (default: False) --work-pool-name: Override work pool name (default: auto-selected based on production flag) --branch: Git branch to deploy from (default: auto-selected based on production flag) --image: Docker image to use (default: auto-selected based on production flag) --optional-tags: Optional deployment tags (space-separated). Environment tag (production/development) is always added automatically. Note: A mandatory environment tag ('production' or 'development') is automatically added based on the --production flag. """ deployer = PrefectFlowDeployer.from_args(python_function, description, optional_tags) try: deployer.deploy() except Exception as e: print(deployer.schedules) raise

Your message was too long, here's your message at the allowed limit: Actually i have been able to override (and inherit) existing schedules for other deploys without slug names and for this one i even tried to give it one in the ui after the issue first appeared Here is my code from typing import Callable import os import inspect import argparse from prefect import flow from prefect client orchestration import SyncPrefectClient from prefect client schemas objects import DeploymentSchedule from prefect exceptions import PrefectHTTPStatusError ObjectNotFound from prefect flows import Flow as FlowClass from prefect_github import GitHubCredentials from prefect runner storage import GitRepository from brrr orchestration import prefect_constants as pfc class PrefectFlowDeployer() def __init__( self python_function Callable production bool=False work_pool_name str=None branch str=None image str = None optional_tags list | str | None = None ) if production work_pool_name = pfc production_pool if not work_pool_name else work_pool_name image = pfc docker_image_repository_prod if not image else image branch = pfc git_prod_branch if not branch else branch else work_pool_name = <http //pfc dev|pfc dev>_pool if not work_pool_name else work_pool_name image = pfc docker_image_repository_dev if not image else image branch = pfc git_dev_branch if not branch else branch self python_function = python_function self work_pool_name = work_pool_name self branch = branch self image = image self production = production self schedules = None self tags = self _create_deployment_tags(production optional_tags) def _create_deployment_name(self) return self input_flow name + '_env_' + ('production' if self production else 'development') + '_branch_' + self branch @staticmethod def _normalize_tags(tags list | str | None) -> list Normalize tags input to a list Args tags Can be None a string or a list of strings Returns A list of strings or empty list if input is None if tags is None return [] elif isinstance(tags str) return [tags] elif isinstance(tags list) return tags def _create_deployment_tags(self production bool optional_tags list | str | None = None) -> list Create deployment tags with mandatory environment tag and optional additional tags Args production Whether this is a production deployment optional_tags Optional additional tags (can be string or list) Returns List of tags including mandatory environment tag and any optional tags # Mandatory environment tag env_tag = 'production' if production else 'development' # Normalize optional tags to list additional_tags = self _normalize_tags(optional_tags) # Combine tags (env tag first then optional tags) return [env_tag] + additional_tags def set_deployment_schedule(self) -> None if self production with SyncPrefectClient(api=pfc prefect_api_url) as client try deployment = client read_deployment_by_name(f {self input_flow name}/{self deployment_name} ) deployment_schedules = client read_deployment_schedules(deployment_id=deployment id) # Extract the actual schedule objects from DeploymentSchedule objects self schedules = [] for deployment_schedule in deployment_schedules if hasattr(deployment_schedule 'schedule') and deployment_schedule schedule self schedules append(deployment_schedule schedule) else # Fallback try to access the schedule directly if it's stored differently for schedule_type in ['cron' 'interval' 'rrule'] schedule = getattr(deployment_schedule schedule_type None) if schedule self schedules append(schedule) break except ObjectNotFound as exc print( Deployment not found in production - no schedule inheritance ) pass def deploy_flow_docker(self) -> None Deploy the flow using the configured settings if not hasattr(self 'input_flow') or not isinstance(self input_flow FlowClass) raise ValueError('input_flow must exist