< Marvin> I have the following setup A docker compose file ` Prefect Community #ask-marvin

<@ULVA73B9P> I have the following setup: A docker...

Martijn Sanderse

09/05/2025, 11:29 AM

@Marvin I have the following setup: A docker compose file:

Copy code

services:

  prefect-server:
    image: prefecthq/prefect:3-latest
    environment:
      PREFECT_RESULTS_DEFAULT_SERIALIZER: json
      PREFECT_RESULTS_PERSIST_BY_DEFAULT: true
      PREFECT_HOME: /data
      PREFECT_RESULTS_LOCAL_STORAGE_PATH: /data/storage
      PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK: local-file-system/default-storage-block
    volumes:
      - /Users/masa/Projects/Prefect/prefect-data:/data  # Persistent storage
    command: prefect server start --host 0.0.0.0
    ports:
      - "4200:4200"

  prefect-worker-1:
    image: prefecthq/prefect:3-latest
    depends_on:
      prefect-server:
        condition: service_started
    environment:
      PREFECT_API_URL: <http://prefect-server:4200/api>
      PREFECT_RESULTS_DEFAULT_SERIALIZER: json
      PREFECT_RESULTS_PERSIST_BY_DEFAULT: true
      PREFECT_HOME: /data
      PREFECT_RESULTS_LOCAL_STORAGE_PATH: /data/storage
      PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK: local-file-system/default-storage-block
    command: prefect worker start --pool docker-pool

A Task/Flow as follows:

Copy code

from prefect import flow, task

@task
def hello_world(name: str):
    message = f"Hello, {name}"
    print(message)
    return message


@flow(name="hello_flow", log_prints=True)
def hello_flow(name: str):
    message = hello_world(name)
    return message

I deployed the workflow like this:

Copy code

# Welcome to your prefect.yaml file! You can use this file for storing and managing
# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: Prefect
prefect-version: 3.4.10

# build section allows you to manage and build docker images
build: null

# push section allows you to manage if and how this project is uploaded to remote locations
push: null

# pull section allows you to provide instructions for cloning this project in remote locations
pull:
- prefect.deployments.steps.git_clone:
    repository: <https://github.com/martijn-sanderse/prefect-demo.git>
    branch: main
    access_token: '{{ prefect.blocks.secret.deployment-hello-deploy-hello-world-flow-repo-token}}'

# the deployments section allows you to provide configuration for deploying flows
deployments:
- name: hello_deploy
  version: null
  tags: []
  concurrency_limit: null
  description: null
  entrypoint: hello.py:hello_flow
  parameters: {}
  work_pool:
    name: docker-pool
    work_queue_name: null
    job_variables: {}
  schedules: []

Finally, I'm using a script to run the deployment

Copy code

from prefect.deployments import run_deployment
from prefect.states import get_state_result

import asyncio


import dotenv
dotenv.load_dotenv() # loads prefect api url from .env


if __name__ == "__main__":

    flow_run = run_deployment(name="hello_flow/hello_deploy", parameters={"name": "Masa"})
    print("result", type(flow_run), type(flow_run.state))
    print ("flow_run", flow_run)

    result = asyncio.run(get_state_result(flow_run.state, raise_on_failure=False))
    print("result", type(result), result)

from the output it seems that the result is stored somewhere (Yay!) But I'm unable to retrieve it. The result I get is:

Copy code

flow_run id=UUID('50d0045d-3099-4d38-bc1a-2ff09a6c6596') name='steadfast-tench' flow_id=UUID('f47ce310-4e48-4640-975c-295cd77faac9') state_id=UUID('0199199f-d73d-76f5-ba43-4dbd6573a207') deployment_id=UUID('80058e5a-ef57-47c3-b26c-c5649060cdba') deployment_version='e85f5bbc' work_queue_name='default' flow_version='8a824ed28ecd42ae153eba02428adc62' parameters={'name': 'Masa'} idempotency_key=None context={} empirical_policy=FlowRunPolicy(max_retries=0, retry_delay_seconds=0.0, retries=0, retry_delay=0, pause_keys=set(), resuming=False, retry_type=None) tags=[] labels={'prefect.flow.id': 'f47ce310-4e48-4640-975c-295cd77faac9', 'prefect.deployment.id': '80058e5a-ef57-47c3-b26c-c5649060cdba', 'prefect.worker.name': 'ProcessWorker 7b1319c6-a8b9-4835-abd9-9a2ae5f65f44', 'prefect.worker.type': 'process', 'prefect.work-pool.name': 'docker-pool', 'prefect.work-pool.id': '8d518e72-a5d3-4fad-afc0-b0ef0d89bc66'} parent_task_run_id=None run_count=1 expected_start_time=DateTime(2025, 9, 5, 11, 25, 25, 138782, tzinfo=Timezone('UTC')) next_scheduled_start_time=None start_time=DateTime(2025, 9, 5, 11, 25, 29, 741435, tzinfo=Timezone('UTC')) end_time=DateTime(2025, 9, 5, 11, 25, 29, 789902, tzinfo=Timezone('UTC')) total_run_time=datetime.timedelta(microseconds=48467) estimated_run_time=datetime.timedelta(microseconds=48467) estimated_start_time_delta=datetime.timedelta(seconds=4, microseconds=602653) auto_scheduled=False infrastructure_document_id=None infrastructure_pid='53' created_by=None work_queue_id=UUID('b3d1ad1c-6b3a-486b-90f7-df7a4ea584d8') work_pool_id=UUID('8d518e72-a5d3-4fad-afc0-b0ef0d89bc66') work_pool_name='docker-pool' state=Completed(message=None, type=COMPLETED, result=ResultRecordMetadata(storage_key='/root/.prefect/storage/080ad9bf7354466bb5696679e40c0510', expiration=None, serializer=PickleSerializer(type='pickle', picklelib='cloudpickle', picklelib_version=None), prefect_version='3.4.14', storage_block_id=None)) job_variables={} state_type=StateType.COMPLETED state_name='Completed'

But when getting the state result, I get a value error: ValueError: Path /root/.prefect/storage/080ad9bf7354466bb5696679e40c0510 does not exist. What I don't understand: • it seems the storage_key is used as a path? But the result is in fact stored somewhere else. • Even though I configured json as default serializer, I'm getting the result serialized by the PickleSerializer. What do I need to change to be able to retrieve the result in my python script and print the contents of the result to the console?

Marvin

09/05/2025, 11:29 AM

✅ thought for 74.3 seconds

Marvin

09/05/2025, 11:30 AM

Thanks! Could you paste the contents of your docker-compose.yml (and any .env you’re using)? Also, a bit more context will help me give you the right advice: - Are you using Prefect Cloud or self-hosting Prefect Server? - Prefect version (assuming 3.x unless you tell me otherwise) - What you want the compose stack to do (e.g., run Prefect Server, start a worker that polls a work pool, run flows in Docker, etc.) - Any errors/logs you’re seeing If it’s helpful, here’s a minimal example for Prefect 3.x that runs a local Prefect Server and a worker that polls a work pool. The worker is configured to talk to the server via PREFECT_API_URL.

Copy code

services:
  prefect-server:
    image: prefecthq/prefect:3-latest
    command: bash -lc "prefect server start"
    ports:
      - "4200:4200"
    restart: unless-stopped

  prefect-worker:
    image: prefecthq/prefect:3-latest
    depends_on:
      - prefect-server
    environment:
      - PREFECT_API_URL=<http://prefect-server:4200/api>
      - PREFECT_LOGGING_LEVEL=INFO
    command: >
      bash -lc "prefect worker start --pool ${PREFECT_WORK_POOL:-default} -n compose-worker"
    restart: unless-stopped

- If you’re using Prefect Cloud, remove the server service and set

PREFECT_API_URL

to your Cloud workspace API URL. - If your deployments use Docker infrastructure, the worker will need access to Docker (e.g., mount the host Docker socket or run Docker-in-Docker), otherwise use Process infrastructure to run flows inside the worker container. Docs that may help while you share details: - Workers and work pools: Workers & Work Pools - Self-hosting Prefect Server: Host Prefect Server - Docker deployments: Docker infrastructure Once you share your compose file and goal, I’ll tailor the setup or debug any issues you’re hitting.

Marvin

09/05/2025, 11:40 AM

Your message was too long, here's your message at the allowed limit: I have the following setup A docker compose file

Copy code

services prefect-server image prefecthq/prefect 3-latest environment PREFECT_RESULTS_DEFAULT_SERIALIZER json PREFECT_RESULTS_PERSIST_BY_DEFAULT true PREFECT_HOME /data PREFECT_RESULTS_LOCAL_STORAGE_PATH /data/storage PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK local-file-system/default-storage-block volumes - /Users/masa/Projects/Prefect/prefect-data /data # Persistent storage command prefect server start --host 0 0 0 0 ports - 4200 4200 prefect-worker-1 image prefecthq/prefect 3-latest depends_on prefect-server condition service_started environment PREFECT_API_URL <http //prefect-server 4200/api> PREFECT_RESULTS_DEFAULT_SERIALIZER json PREFECT_RESULTS_PERSIST_BY_DEFAULT true PREFECT_HOME /data PREFECT_RESULTS_LOCAL_STORAGE_PATH /data/storage PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK local-file-system/default-storage-block command prefect worker start --pool docker-pool

A Task/Flow as follows

Copy code

from prefect import flow task @task def hello_world(name str) message = f Hello {name} print(message) return message @flow(name= hello_flow log_prints=True) def hello_flow(name str) message = hello_world(name) return message

I deployed the workflow like this

Copy code

# Welcome to your prefect yaml file! You can use this file for storing and managing # configuration for deploying your flows We recommend committing this file to source # control along with your flow code # Generic metadata about this project name Prefect prefect-version 3 4 10 # build section allows you to manage and build docker images build null # push section allows you to manage if and how this project is uploaded to remote locations push null # pull section allows you to provide instructions for cloning this project in remote locations pull - prefect deployments steps git_clone repository <https //github com/martijn-sanderse/prefect-demo git> branch main access_token '{{ prefect blocks secret deployment-hello-deploy-hello-world-flow-repo-token}}' # the deployments section allows you to provide configuration for deploying flows deployments - name hello_deploy version null tags [] concurrency_limit null description null entrypoint hello py hello_flow parameters {} work_pool name docker-pool work_queue_name null job_variables {} schedules []

Finally I'm using a script to run the deployment

Copy code

from prefect deployments import run_deployment from prefect states import get_state_result import asyncio import dotenv dotenv load_dotenv() # loads prefect api url from env if __name__ == __main__ flow_run = run_deployment(name= hello_flow/hello_deploy parameters={ name Masa }) print( result type(flow_run) type(flow_run state)) print ( flow_run flow_run) result = asyncio run(get_state_result(flow_run state raise_on_failure=False)) print( result type(result) result)

from the output it seems that the result is stored somewhere (Yay!) But I'm unable to retrieve it The result I get is ```flow_run id=UUID('50d0045d-3099-4d38-bc1a-2ff09a6c6596') name='steadfast-tench' flow_id=UUID('f47ce310-4e48-4640-975c-295cd77faac9') state_id=UUID('0199199f-d73d-76f5-ba43-4dbd6573a207') deployment_id=UUID('80058e5a-ef57-47c3-b26c-c5649060cdba') deployment_version='e85f5bbc' work_queue_name='default' flow_version='8a824ed28ecd42ae153eba02428adc62' parameters={'name' 'Masa'} idempotency_key=None context={} empirical_policy=FlowRunPolicy(max_retries=0 retry_delay_seconds=0 0 retries=0 retry_delay=0 pause_keys=set() resuming=False retry_type=None) tags=[] labels={'prefect flow id' 'f47ce310-4e48-4640-975c-295cd77faac9' 'prefect deployment id' '80058e5a-ef57-47c3-b26c-c5649060cdba' 'prefect worker name' 'ProcessWorker 7b1319c6-a8b9-4835-abd9-9a2ae5f65f44' 'prefect worker type' 'process' 'prefect work-pool name' 'docker-pool' 'prefect work-pool id' '8d518e72-a5d3-4fad-afc0-b0ef0d89bc66'} parent_task_run_id=None run_count=1 expected_start_time=DateTime(2025 9 5 11 25 25 138782 tzinfo=Timezone

Marvin

09/05/2025, 11:40 AM

('UTC')) next_scheduled_start_time=None start_time=DateTime(2025 9 5 11 25 29 741435 tzinfo=Timezone('UTC')) end_time=DateTime(2025 9 5 11 25 29 789902 tzinfo=Timezone('UTC')) total_run_time=datetime timedelta(microseconds=48467) estimated_run_time=datetime timedelta(microseconds=48467) estimated_start_time_delta=datetime timedelta(seconds=4 microseconds=602653) auto_scheduled=False infrastructure_document_id=None infrastructure_pid='53' created_by=None work_queue_id=UUID('b3d1ad1c-6b3a-486b-90f7-df7a4ea584d8') work_pool_id=UUID('8d518e72-a5d3-4fad-afc0-b0ef0d89bc66') work_pool_name='docker-pool' state=Completed(message=None type=COMPLETED result=ResultRecordMetadata(storage_key='/root/ prefect/storage/080ad9bf7354466bb5696679e40c0510' expiration=None serializer=PickleSerializer(type='pickle' picklelib='cloudpickle' picklelib_version=None) prefect_version='3 4 14' storage_block_id=None)) job_variables={} state_type=StateType COMPLETED state_name='Completed'``` But when getting the state result I get a value error ValueError Path /root/ prefect/storage/080ad9bf7354466bb5696679e40c0510 does not exist What I don't understand • it seems the storage_key is used as a path? But the result is in fact stored somewhere else • Even though I

Open in Slack

Previous Next