<@ULVA73B9P> Can I create a deployment from a flow...
# ask-marvin
s
@Marvin Can I create a deployment from a flow defined within a class?
m
thought for 297.3 seconds
Short answer: yes. In Prefect 3.x you can define a flow as a method on a class and create a deployment from it. The supported and recommended patterns are class methods and static methods; instance methods are not suitable for deployments because there’s no way for Prefect to construct your instance at run time. What works - Static method
Copy code
from prefect import flow, get_run_logger

class MyFlows:
    @flow
    @staticmethod
    def my_flow(name: str = "world"):
        logger = get_run_logger()
        <http://logger.info|logger.info>(f"Hello, {name}")
        return name
- Class method
Copy code
from prefect import flow

class MyFlows:
    @flow
    @classmethod
    def my_flow(cls, x: int = 1):
        return x + 1
Key tips - Put @flow outermost: use
@flow
above
@staticmethod
or
@classmethod
. Prefect’s
Flow
decorator accepts
staticmethod
and
classmethod
objects. - Your entrypoint must be importable in the execution environment and can reference class methods with a dotted path like
path/to/file.py:MyFlows.my_flow
. - Avoid instance methods that require
self
state; deployments cannot serialize or reconstruct an instance. If you need configuration, pass it as parameters or pull it from blocks/env. How to deploy - CLI (from your project root):
Copy code
prefect deploy ./flows.py:MyFlows.my_flow -n my-deployment
- Programmatically from source (e.g., Git):
Copy code
from prefect import Flow

Flow.from_source(
    "<https://github.com/you/your-repo>",
    entrypoint="flows.py:MyFlows.my_flow",
).deploy(name="my-deployment", work_pool_name="process")
References - Create deployments (how-to) - Flow API reference - Entrypoint type (importability rules) If you share a small snippet of your class/flow, I can suggest the exact entrypoint and deployment command.
s
@Marvin when I do
<http://flow.to|flow.to>_deployment(entrypoint_type=EntrypointType.MODULE_PATH)
with a docker work pool, I see "prefect.exceptions.MissingFlowError: Flow function with name '<flow-name>' not found in '<module>'.
m
thought for 329.2 seconds
Thanks for the details — that error usually means Prefect could import your module, but could not find the specified flow function inside it. When you set
entrypoint_type=EntrypointType.MODULE_PATH
, Prefect expects an importable module path and a Python attribute path (not the flow’s display name). For class-based flows, that means the right-hand side must include the class, e.g.
your_pkg.your_mod:MyFlows.my_flow
. Common causes and fixes - Wrong attribute path - If your flow is a class method or static method, the entrypoint must be
module:ClassName.method_name
— not just
module:method_name
. - Custom flow name vs. function name - If you used
@flow(name="pretty-name")
, remember the entrypoint still uses the Python attribute path, not the flow’s display name. Use
my_module:MyFlows.my_flow
, not the custom name. - Decorator order on class methods - Use
@flow
outermost:
Copy code
class MyFlows:
        @flow
        @classmethod
        def my_flow(cls): ...
or
Copy code
class MyFlows:
        @flow
        @staticmethod
        def my_flow(): ...
- Module is importable locally but symbol isn’t exposed - Ensure the flow is defined at import time (not under
if __name__ == "__main__":
) and that the symbol is accessible via the module path you’re using. - Docker + MODULE_PATH specifics - MODULE_PATH requires your code to be importable inside the Docker image (installed package or added to
PYTHONPATH
). If your image doesn’t contain your project as an installed module, prefer the default script-path entrypoint or build an image that installs your package. Quick local sanity check - Inspect what Prefect will use:
Copy code
print(my_flow.fn.__module__)
  print(my_flow.fn.__qualname__)  # e.g., "MyFlows.my_flow" for class methods
- Test resolution locally:
Copy code
from prefect.flows import safe_load_flow_from_entrypoint
  ep = f"{my_flow.fn.__module__}:{my_flow.fn.__qualname__}"
  print("Testing entrypoint:", ep)
  safe_load_flow_from_entrypoint(ep)  # should return a Flow or raise with a precise reason
For Docker work pools - If you want to stick with MODULE_PATH, make sure your image installs your project (e.g.,
pip install .
) so
import your_pkg.your_mod
works in the container. - Otherwise, use the script-path entrypoint (default) or call
flow.deploy(image=..., work_pool_name="docker", build=True, push=True)
so Prefect builds an image that includes your code. Docs - Create deployments - Flow API reference - Entrypoint type (import rules) If you share: - the module path you expect to use, - the class/method signature and decorator order, - and whether your project is installed in the Docker image, I can pinpoint the exact entrypoint string and the minimal changes needed.
s
@Marvin the qualname resolves to the proper Class.flow_name, but it looks like resolution with class and static methods only works when using path. Here is the docstring for the function you recommended:
Copy code
def safe_load_flow_from_entrypoint(entrypoint: str) -> Optional[Flow[P, Any]]:
    """
    Safely load a Prefect flow from an entrypoint string. Returns None if loading fails.

    Args:
        entrypoint (str): A string identifying the flow to load. Can be in one of the following formats:
            - `<path_to_script>:<flow_func_name>`
            - `<path_to_script>:<class_name>.<flow_method_name>`
            - `<module_path>.<flow_func_name>`

    Returns:
        Optional[Flow]: The loaded Prefect flow object, or None if loading fails due to errors
        (e.g. unresolved dependencies, syntax errors, or missing objects).
    """
When I try to use EntrypointType.FILE_PATH with
<http://my_flow.to|my_flow.to>_deployment()
, I get the following
Copy code
File "[...]/prefect/flows.py", line 960, in to_deployment
    return RunnerDeployment.from_flow(
  File "[...]/prefect/deployments/runner.py", line 769, in from_flow
    Path(flow_file).absolute().relative_to(Path.cwd().absolute())
  File "[...]/pathlib.py", line 818, in relative_to
    raise ValueError("{!r} is not in the subpath of {!r}"
ValueError: '<absolute path of flow>.py' is not in the subpath of '<absolute path of where I am running the code>' OR one path is relative and the other is absolute.
Is there a proper way to define a deployment with to_deployment() such that it can find the flow file?
m
thought for 56.8 seconds
Great debugging — you’re right about the current behavior. - EntrypointType.MODULE_PATH: today it only reliably resolves top-level flow functions (module_attr), not class/staticmethods. That’s why you’re seeing MissingFlowError when using a class/staticmethod with MODULE_PATH. - EntrypointType.FILE_PATH: supports class/staticmethods via path.py:ClassName.method_name. Your ValueError indicates Prefect is trying to make the flow’s file path relative to your current working directory — but your CWD isn’t the project root containing that file. How to fix and options 1) Use FILE_PATH and run from project root - Run your code (or
prefect deploy
) from the project root (where prefect.yaml lives) so the flow file is a subpath of CWD. Otherwise you’ll hit: ValueError: '<flow>.py' is not in the subpath of '<cwd>' - Quick guard:
Copy code
import os, inspect
  from pathlib import Path

  # Ensure we’re at project root
  print("CWD:", Path.cwd())
  from mypkg.flows import MyFlows
  print("Flow file:", Path(inspect.getsourcefile(MyFlows.my_flow)).resolve())
- Then:
Copy code
from prefect.types.entrypoint import EntrypointType

  MyFlows.my_flow.to_deployment(
      name="my-deployment",
      work_pool_name="docker",
      image="<http://ghcr.io/you/your-image:latest|ghcr.io/you/your-image:latest>",
      build=True,
      push=True,
      entrypoint_type=EntrypointType.FILE_PATH,
  )
- For Docker, FILE_PATH works fine; Prefect will package/build the image when you pass image/build/push. 2) If you must use MODULE_PATH with a class/staticmethod - Promote the method to a module-level alias so MODULE_PATH can resolve it:
Copy code
# mypkg/flows.py
  from prefect import flow

  class MyFlows:
      @flow
      @staticmethod
      def my_flow(...):
          ...

  # Module-level alias for MODULE_PATH entrypoint support
  my_flow = MyFlows.my_flow
- Then your MODULE_PATH entrypoint is
mypkg.flows:my_flow
. Also ensure your code is importable in the Docker image (e.g., pip install . in the image). 3) Alternative: top-level wrapper - Wrap the class method in a top-level flow and delegate:
Copy code
from prefect import flow

  @flow
  def my_flow(...):
      return MyFlows.my_flow(...)
Now
mypkg.flows:my_flow
works with MODULE_PATH. 4) Or avoid local path resolution entirely - Use Flow.from_source with an explicit entrypoint and let Prefect manage packaging:
Copy code
from prefect import Flow

  Flow.from_source(
      "<https://github.com/you/your-repo>",
      entrypoint="flows.py:MyFlows.my_flow",  # class/staticmethod ok here
  ).deploy(name="my-deployment", work_pool_name="docker")
Why the errors happened - MissingFlowError with MODULE_PATH: loader looks for a top-level attribute in the module; class/staticmethods aren’t supported there right now. - ValueError with FILE_PATH: CWD must be the project root so Prefect can write a relative entrypoint in the deployment. Running from outside the project root triggers that pathlib relative_to error. Docs - Create deployments - Entrypoint type (import rules) - Flow API reference If you share your project layout and where you’re calling to_deployment from (CWD vs. flow file location), I can suggest the exact fix (chdir, alias, or wrapper) and the entrypoint string to use.
s
@Marvin Changing my working directory when running deploy() did resolve that relative path error error but I now get this:
Copy code
Unexpected exception encountered when trying to load flow
Traceback (most recent call last):
  File "/app/lib/python3.10/site-packages/prefect/engine.py", line 112, in <module>
    flow: "Flow[..., Any]" = load_flow(flow_run)
  File "/app/lib/python3.10/site-packages/prefect/flow_engine.py", line 150, in load_flow
    flow = run_coro_as_sync(
  File "/app/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 207, in run_coro_as_sync
    return call.result()
  File "/app/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 365, in result
    return self.future.result(timeout=timeout)
  File "/app/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 192, in result
    return self.__get_result()
  File "/app/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/app/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 441, in _run_async
    result = await coro
  File "/app/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 188, in coroutine_wrapper
    return await task
  File "/app/lib/python3.10/site-packages/prefect/client/utilities.py", line 69, in wrapper
    return await func(client, *args, **kwargs)
  File "/app/lib/python3.10/site-packages/prefect/flows.py", line 2806, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=from_path, local_path=".")
  File "/app/lib/python3.10/site-packages/prefect/filesystems.py", line 162, in aget_directory
    copytree(from_path, local_path, dirs_exist_ok=True, ignore=ignore_func)
  File "/app/lib/python3.10/shutil.py", line 557, in copytree
    with os.scandir(src) as itr:
FileNotFoundError: [Errno 2] No such file or directory: '/None'
m
thought for 66.3 seconds
Thanks — that traceback points to Prefect trying to fetch your project files from “storage” at run time, but the deployment doesn’t have a valid source path configured, so it ends up as from_path="/None". Why it happens - With docker work pools, your flow code must be available inside the container either: - baked into the image, or - downloadable from a configured storage/source (e.g., Git repo, S3/GCS, etc.). - Using
to_deployment()
with
entrypoint_type=FILE_PATH
does not build an image or attach storage by itself. If you don’t provide an image or a source, the runner still tries to fetch code and hits from_path=None → FileNotFoundError: '/None'. Fix options A) Bake your code into an image (recommended for docker pools) - Use
deploy()
so Prefect can build/push your image and avoid storage fetches:
Copy code
from prefect.types.entrypoint import EntrypointType

MyFlows.my_flow.deploy(
    name="my-deployment",
    work_pool_name="docker",
    image="<http://ghcr.io/you/your-app:latest|ghcr.io/you/your-app:latest>",
    build=True,
    push=True,
    entrypoint_type=EntrypointType.FILE_PATH,  # supports Class.method
)
- Run this from your project root (where prefect.yaml is) so the file path is relative. - At run time, the container already has your code; Prefect won’t try to download from storage. B) Keep using to_deployment(), but attach a source - Define the flow from a source so the deployment carries storage info:
Copy code
from prefect import Flow

flow = Flow.from_source(
    "<https://github.com/you/your-repo>",  # or a filesystems block-backed source
    entrypoint="flows.py:MyFlows.my_flow",
)
flow.to_deployment(
    name="my-deployment",
    work_pool_name="docker",
    job_variables={"image": "prefecthq/prefect:3-python3.10"},  # the runtime image
)
- Here Prefect will clone/download the source into the container before loading your flow. C) If you must use MODULE_PATH - Alias the class/staticmethod to a top-level symbol so module-path resolution works:
Copy code
class MyFlows:
    @flow
    @staticmethod
    def my_flow(...):
        ...

my_flow = MyFlows.my_flow  # module-level alias
- Then deploy with MODULE_PATH, but you still need either (A) an image with code baked in or (B) a source attached. Quick checks - In the UI, open the deployment and check “Storage/Source”. If it’s None, your image must contain your code. - Ensure your docker work pool has an image set or you pass one via
deploy(image=...)
or
to_deployment(job_variables={"image": ...})
. Docs - Create deployments - Flow API reference If you share: - which of
to_deployment()
vs
deploy()
you’re calling, - your project layout, - and whether you want code baked into the image or pulled from a repo, I’ll suggest the exact invocation (and Dockerfile snippet if needed) to eliminate the '/None' error.
s
@Marvin I now have flows of static and class methods working. Can subclasses leverage the superclass's flow class method but provide overriding class variables? I'd like to do something like this and have it print super or sub accordingly Example:
Copy code
class MyFlows:
    my_var = "super"

    @flow
    @classmethod
    def my_flow(cls, ...):
        print(cls.my_var)

class MySubClass(MyFlows):
    my_var = "sub"

deploy(MySubClass.my_flow.to_deployment(...))
m
thought for 408.0 seconds
Short answer: Yes. Prefect 3.x supports classmethod flows, and subclass overrides will be respected. Accessing the flow via the subclass will bind
cls
to the subclass, so
cls.my_var
prints the overridden value. Important: Use the decorators in this order —
@flow
outermost,
@classmethod
inner. Example:
Copy code
from prefect import flow

class MyFlows:
    my_var = "super"

    @flow
    @classmethod
    def my_flow(cls):
        print(cls.my_var)

class MySubClass(MyFlows):
    my_var = "sub"

MyFlows.my_flow()      # prints "super"
MySubClass.my_flow()   # prints "sub"
Deploying in Prefect 3.x - Directly deploy to a work pool:
Copy code
MySubClass.my_flow.deploy(
    name="my-sub-flow",
    work_pool_name="my-pool",
)
- Or “serve” locally (no work pool) with a schedule:
Copy code
MySubClass.my_flow.serve(
    name="my-sub-flow",
    cron="0 * * * *",
)
- If you prefer
to_deployment
+ batch deploy:
Copy code
from prefect.deployments import deploy

d = MySubClass.my_flow.to_deployment(name="my-sub-flow")
deploy(d, work_pool_name="my-pool")
- For remote source code:
Copy code
from prefect import Flow

Flow.from_source(
    "<https://github.com/owner/repo.git>",
    "path/to/file.py:MyFlows.my_flow"
).deploy(
    "my-sub-flow",
    work_pool_name="my-pool",
)
Notes - Avoid removed 2.x patterns like
Deployment.build_from_flow()
or
prefect deployment build
. - Your
deploy(MySubClass.my_flow.to_deployment(...))
pattern is valid in 3.x; alternatively, prefer
MySubClass.my_flow.deploy(...)
for simplicity. Docs - Deploy via Python API - Create deployments overview - Run flows in local processes (serve)
s
@Marvin
Copy code
print(MySubClass.my_flow.__module__)
print(MySubClass.my_flow.__qualname__)
and
Copy code
print(MyFlows.my_flow.__module__)
print(MyFlows.my_flow.__qualname__)
produce the same output, namely the superclass module and MyFlows.my_flow respectively. Are you sure a deployment would respect the class variables from MySubClass?
m
thought for 436.8 seconds
Great question — and your observation is correct: in Python,
__module__
and
__qualname__
on a method reflect where the function was defined (the base class), not how it was accessed. So seeing
MyFlows.my_flow
for both is expected. What matters for deployments is the entrypoint string the deployment stores. That’s what the worker will resolve at run time, and it determines which class “owns” the flow at binding time. How to verify what your deployment will do - Inspect the entrypoint on the deployment you create:
Copy code
d = MySubClass.my_flow.to_deployment("test-sub")
print(d.entrypoint)
If you see something like: - module:MySubClass.my_flow — runs will bind via the subclass and
cls.my_var
will be "sub" - module:MyFlows.my_flow — runs will bind via the base class and
cls.my_var
will be "super" You can also see the entrypoint in the Deployment details in the UI. If the entrypoint shows the base class If your deployment shows
...:MyFlows.my_flow
, the safest way to ensure subclass behavior is to give the subclass its own (thin) classmethod wrapper so the entrypoint is explicitly the subclass:
Copy code
from prefect import flow

class MyFlows:
    my_var = "super"

    @flow
    @classmethod
    def my_flow(cls):
        print(cls.my_var)

class MySubClass(MyFlows):
    my_var = "sub"

    @flow
    @classmethod
    def my_flow(cls):
        # delegate to the base implementation
        return super().my_flow()
Now deploy via the subclass:
Copy code
MySubClass.my_flow.deploy(
    name="my-sub-flow",
    work_pool_name="my-pool",
)
That guarantees the deployment entrypoint is
...:MySubClass.my_flow
and runs will print "sub". Alternative: if you deploy from a remote source, you can explicitly set the entrypoint to the subclass:
Copy code
from prefect import Flow

Flow.from_source(
    "<https://github.com/owner/repo.git>",
    "path/to/file.py:MySubClass.my_flow",
).deploy(
    "my-sub-flow",
    work_pool_name="my-pool",
)
Would you be able to print the
d.entrypoint
from your current code? If it shows the base class, use the thin override above to force subclass binding. Docs for reference - Deploy via Python API - Create deployments overview
s
@Marvin printing d.entrypoint for a deployment from the subclass produces the superclass as an entrypoint
m
thought for 104.3 seconds
Thanks for checking — you’re right. Since the deployment’s entrypoint is derived from the function’s definition site (
__qualname__
), it points to
MyFlows.my_flow
. When the worker loads that entrypoint, it accesses the descriptor on
MyFlows
, so the flow binds
cls
to
MyFlows
(not
MySubClass
). In other words: a deployment created from
MySubClass.my_flow
will still run with
cls = MyFlows
if the entrypoint is
...:MyFlows.my_flow
. Ways to ensure the subclass is respected at runtime 1) Recommended: give the subclass a thin overriding classmethod flow so the entrypoint resolves to the subclass.
Copy code
from prefect import flow

class MyFlows:
    my_var = "super"

    @flow
    @classmethod
    def my_flow(cls):
        print(cls.my_var)

class MySubClass(MyFlows):
    my_var = "sub"

    @flow
    @classmethod
    def my_flow(cls):
        # delegate to the base implementation
        return super().my_flow()
Then deploy via the subclass so the entrypoint is `...:MySubClass.my_flow`:
Copy code
MySubClass.my_flow.deploy(
    name="my-sub-flow",
    work_pool_name="my-pool",
)
2) Use the CLI with an explicit entrypoint to the subclass In prefect.yaml:
Copy code
deployments:
- name: my-sub-flow
  entrypoint: path/to/file.py:MySubClass.my_flow
  work_pool:
    name: my-pool
Then:
Copy code
prefect deploy
3) Use Flow.from_source with an explicit entrypoint If you deploy from source, set the entrypoint to the subclass:
Copy code
from prefect import Flow

Flow.from_source(
    "<https://github.com/owner/repo.git>",
    "path/to/file.py:MySubClass.my_flow",
).deploy(
    "my-sub-flow",
    work_pool_name="my-pool",
)
Key takeaway - The entrypoint controls which class the descriptor binds to at runtime. If it’s
MyFlows.my_flow
,
cls
will be
MyFlows
. To get
cls = MySubClass
, the entrypoint must be
MySubClass.my_flow
. Related docs - Deploy via Python API - Create deployments overview - Load flow from entrypoint If you share your project layout, I can suggest the cleanest way to structure the override.