flow.run_config

flow.storage

flow.schedule

deploy.py

"""Prefect Flow Deployment Recipe."""
import os
from datetime import timedelta

from prefect.run_configs import KubernetesRun
from prefect.schedules import IntervalSchedule
from prefect.storage import Docker

from src import config
from src.flow import flow

# Setting flow run config with `prefect.run_config`
flow.run_config = KubernetesRun(
    job_template_path=config.KUBE_JOB_TEMPLATE_PATH,
    cpu_limit=config.KUBE_CPU_LIMIT,
    cpu_request=config.KUBE_CPU_REQUEST,
    memory_limit=config.KUBE_MEMORY_LIMIT,
    memory_request=config.KUBE_MEMORY_REQUEST,
)

# Setting flow storage with `prefect.storage`
flow.storage = Docker(
    env_vars=config.ENVIRONMENT_VARIABLES,
    dockerfile="Dockerfile",
    image_name=config.DOCKER_IMAGE_NAME,
    image_tag=config.DOCKER_IMAGE_TAG,
    registry_url=config.DOCKER_REGISTRY_URL,
    build_kwargs={
        "buildargs": {
            "PIP_INDEX_URL": os.getenv("PIP_INDEX_URL"),
        },
        "nocache": True,
    },
)

# Setting flow schedule with `prefect.schedules`
# NOTE: Make sure to import the desired schedule type.
flow.schedule = IntervalSchedule(interval=timedelta(hours=1))

"""Prefect Flow Deployment Recipe."""
import yaml

from prefect.deployments import Deployment
from prefect.packaging import DockerPackager
from prefect.infrastructure import KubernetesJob

from src import config
from src.flow import flow

packager = DockerPackager(
    dockerfile="Dockerfile",
    registry_url=config.DOCKER_REGISTRY_URL
),

deployment = Deployment(
    flow=flow,
    name="sandbox",
    infrastructure=KubernetesJob(
        image="?",  # I would rather not pass this as a static value
        job=yaml.safe_load(config.KUBE_JOB_TEMPLATE_PATH),
    )
)

schedule = "Can only be defined in a deployment.yaml?"

KubernetesJob