import logging
import datetime
from prefect.deployments import Deployment
from prefect.orion.schemas.schedules import IntervalSchedule
from dataclasses import asdict

from precog_api.prefect.flows.load_data_from_minio_flow import load_data_from_minio
from precog_api.config import Config
from prefect.filesystems import RemoteFileSystem
from prefect.infrastructure import DockerContainer
from prefect.infrastructure.docker import DockerRegistry

registry_block = DockerRegistry.load("gitlab-docker-registry")

storage_block = RemoteFileSystem(
    basepath="<s3://flow-storage/load_data_from_minio>",
    settings={
        "use_ssl": False,
        "key": Config.MINIO_USER,
        "secret": Config.MINIO_PASSWORD,
        "client_kwargs": {"endpoint_url": f"http://{Config.MINIO_ENDPOINT}"},
    },
)

storage_block.save("prefect-load-data-from-minio", overwrite=True)

infra_block = DockerContainer(
    registry=registry_block,
    image="<http://registry.gitlab.com/periplo-innovation/project-precog/ml_db|registry.gitlab.com/periplo-innovation/project-precog/ml_db>",
    image_pull_policy="IF_NOT_PRESENT",
    networks=[Config.PREFECT_CONTAINER_NETWORK]
    if Config.PREFECT_CONTAINER_NETWORK
    else None,
    env=asdict(Config()),
)
infra_block.save("precog-api-docker", overwrite=True)

today = datetime.datetime.today()

schedule = IntervalSchedule(
    interval=datetime.timedelta(days=1),
    anchor_date=datetime.datetime(
        year=today.year,
        month=today.month,
        day=today.day,
        hour=1,
    ),  # Run every day at 1 AM
)


deployment = Deployment.build_from_flow(
    flow=load_data_from_minio,
    name="load_data_from_minio",
    work_queue_name="default",
    schedule=schedule,
    storage=storage_block,
    infrastructure=infra_block,
)

if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    deployment.apply()

httpx.HTTPStatusError: Client error '422 Unprocessable Entity' for url '<http://prefect-server:4200/api/deployments/>'
For more information check: <https://httpstatuses.com/422>

PrefectHTTPStatusError

apply

if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    print("Upload to stroage")
    deployment.upload_to_storage()
    print("Apply")
    deployment.apply(upload=False)