Thanks for the detailed context — this is a great write-up. You’ve run into a limitation of the experimental bundles used by infrastructure-bound flows (e.g.,

@kubernetes

) in Prefect 3.x: when Prefect registers “local” modules for pickle-by-value, any classes/functions defined in those modules are serialized by value with cloudpickle. Django model modules that reference cryptography objects with thread locks are a classic case that will fail with “cannot pickle ‘_thread.lock’ object”. Direct answer - There isn’t a public setting, decorator option, or env var today to: - exclude specific modules from bundle serialization, or - force pickle-by-reference globally for all local modules - The

@kubernetes

(and friends) decorators do not expose serialization controls; bundles are always used for ad‑hoc submissions. References: - Submit flows directly to dynamic infrastructure - Settings reference (no bundle/serialization flags) Practical ways to get unblocked Pick one of these depending on your constraints. A) Move away from ad-hoc bundles: use a deployment with source or a container image - With a deployment, your code is pulled by reference (from a repo or image), not serialized by value, so Django + crypto objects won’t be pickled. - Two easy paths: 1) Python API

from prefect import flow, Flow

  @flow
  async def process_item(data):
      ...

  # Point to your repo and entrypoint; the image is the one your K8s worker will use
  Flow.from_source(
      "<https://github.com/your-org/your-repo>",
      "src/myproj/flows.py:process_item",
  ).deploy(
      name="process-item",
      work_pool_name="my-k8s-pool",
      job_variables={"image": "<http://ghcr.io/your-org/app:tag|ghcr.io/your-org/app:tag>"},
  )

2) CLI (verify options with

prefect deploy --help

)

prefect deploy ./src/myproj/flows.py:process_item -n process-item -p my-k8s-pool

Then configure the work pool’s base job template/image or pass job variables so the worker runs with an image that contains your Django app and dependencies. Docs: - Deployment overview B) Keep

@kubernetes

, but make your Django app a “third-party” package (avoid being treated as local) - Package your Django code as a non-editable wheel (not

-e

). Install it into: - the environment where you submit the flow, and - the image used by your Kubernetes work pool. - When Prefect sees your package import from site-packages (instead of your working directory), it won’t register it as a “local module” for pickle-by-value, so your Django models won’t be serialized. - Use a custom image via the work pool or the decorator’s job variables:

@kubernetes(work_pool="my-k8s-pool", image="<http://ghcr.io/your-org/app:tag|ghcr.io/your-org/app:tag>")
  @flow
  async def process_item(data):
      ...

- In practice: - Build wheel:

python -m build

(or your preferred tool) - Install non-editable:

pip install dist/myapp-*.whl

in both environments - Confirm in logs that

myapp.*

no longer shows up under “Registering local modules for pickle-by-value” C) Reduce what cloudpickle needs to capture (works sometimes, but not always enough) Even with ad-hoc/bundles, you can sometimes avoid pulling in the Django model classes during pickling: - Do not reference Django models (or crypto objects) at function definition time: - no module-level imports that set globals used by tasks/flows - no default argument values that are model instances/classes - avoid type hints that directly reference Django model classes; prefer strings or

Any

- Import models inside the function body and keep them in local variables: ``` from future import annotations # to keep annotations as strings @task def save_to_database_task(data, result): # Local import so the function's global namespace does not include model classes

from myapp.database import models as m # Avoid type annotations referencing m.MyModel at def time m.MyModel.objects.create(...) ``` - Ensure no global singletons/ciphers are constructed at import time. However, given your logs show Prefect is registering

myapp.database.models

for by-value, and Django models often carry non-picklable state, C may not be sufficient on its own. What I recommend for your case - If you can: switch this flow to a Deployment (A). It’s the most robust path and avoids bundles entirely. - If you need to keep ad-hoc submission: package your Django app (B) and run with a custom image that has it installed. This prevents Prefect from treating your app as “local” and trying to pickle classes from it. If you want, share: - how you’re running the submitter (is

myapp

installed editably there?), and - your Kubernetes work pool image setup. I can help you pick the cleanest path and provide exact commands for your setup. Useful links - Submit flows directly to dynamic infrastructure - Deployment overview - prefect-kubernetes integration

✅ thought for 291.4 seconds

You can get exactly “one pod per flow run” with the standard KubernetesWorker + deployments flow in Prefect 3.x — no decorators or bundles required. The worker is just a lightweight poller; for each scheduled flow run it submits a separate Kubernetes Job/Pod that executes your flow code. Your worker does not execute the flow itself. Recommended setup - Create a Kubernetes work pool

prefect work-pool create "my-k8s-pool" --type kubernetes

Tip: Inspect the pool’s base job template to see which job_variables you can set (image, namespace, env, resources, etc.), or customize it:

prefect work-pool get-default-base-job-template --type kubernetes > base-job-template.json
  prefect work-pool create "my-k8s-pool" --type kubernetes --base-job-template ./base-job-template.json --overwrite

- Package your Django app into a container image (non-editable install) so your code is imported by reference in the runtime pod (avoids bundles/pickling entirely). Push it to your registry. - Create a Deployment that points to your repo entrypoint and uses the image/job_variables for K8s

from prefect import flow, Flow

  @flow
  async def process_item(data):
      ...

  Flow.from_source(
      "<https://github.com/your-org/your-repo>",
      "src/myproj/flows.py:process_item",
  ).deploy(
      name="process-item",
      work_pool_name="my-k8s-pool",
      job_variables={
          "image": "<http://ghcr.io/your-org/app:tag|ghcr.io/your-org/app:tag>",
          # variable names must match your work pool's base job template
          # examples commonly supported by the k8s template:
          "namespace": "prefect-flows",
          "env": {"DJANGO_SETTINGS_MODULE": "myproj.settings"},
          # many templates expose a 'resources' variable you can use:
          "resources": {
              "requests": {"cpu": "500m", "memory": "1Gi"},
              "limits": {"cpu": "2", "memory": "4Gi"},
          },
      },
  )

- Start a Kubernetes worker to poll the pool

prefect worker start --pool my-k8s-pool

Per-run resource isolation and overrides - Every flow run will create its own Kubernetes Job/Pod from the pool’s template, so you get isolation and per-run CPU/memory settings. - To change resources per execution, pass job variable overrides when you trigger a run:

prefect deployment run process-item/process-item \
    -jv resources='{"requests":{"cpu":"1","memory":"2Gi"},"limits":{"cpu":"3","memory":"6Gi"}}' \
    -p data='{"id": 123}'

The exact job_variables keys you can set come from your pool’s base job template — use:

prefect work-pool inspect my-k8s-pool

to see the schema and the variables section; adjust your overrides accordingly. Why this meets your requirement - The Kubernetes worker submits one Job/Pod per flow run using the template — the worker pod only orchestrates, it does not run user code. - Using deployments with your code in an image avoids the ad‑hoc bundle serializer entirely, so Django + cryptography objects won’t be pickled. Helpful docs - Kubernetes integration: prefect-kubernetes - Work pools/workers: Work pools and Workers - Create/customize deployments: Create deployments and Run on Kubernetes - Job variables/overrides: Customize job variables If you want, share your pool’s base job template (or run

prefect work-pool inspect my-k8s-pool --output json

) and I’ll point out the exact variable names to use for resources/env/image.

✅ thought for 89.9 seconds

You’re super close — the error is coming from using .aio on a classmethod. Why it fails -

Flow.from_source

is a classmethod. The

.aio

wrapper works for functions/methods, but on classmethods it loses the implicit class binding — hence “Flow.afrom_source() missing 1 required positional argument: 'cls'”. Fixes Use the sync methods directly (safe inside async functions), or only use

.aio

on instance methods. Option A: Keep everything synchronous (simplest and recommended)

from pathlib import Path
from prefect.flows import Flow

async def create_deployment_if_needed():
    try:
        flow_obj = Flow.from_source(
            source=str(Path(__file__).parent.parent.parent.parent.parent),
            # MUST be "./path/to/file.py:flow_func_name"
            entrypoint="path/to/my/flow_file.py:process_item",
        )

        deployment_id = flow_obj.deploy(
            name="my-new-deployment",
            work_pool_name="work-pool-1",
            job_variables={
                "image": "my-docker-image:latest",
                "namespace": "my-namespace",
                "image_pull_policy": "Always",
            },
            build=False,   # assumes your image already contains your code/deps
            push=False,    # set True if you are building/pushing with Prefect
            print_next_steps=False,
        )
        print(f":white_check_mark: Created/updated deployment with ID: {deployment_id}")
    except Exception as e:
        import traceback; traceback.print_exc()
        print(f":warning: Deployment creation failed: {e}")
        raise

Option B: If you really want async wrappers, use them only on the instance method

flow_obj = Flow.from_source(
    source="...",
    entrypoint="path/to/my/flow_file.py:process_item",
)
deployment_id = await flow_obj.deploy.aio(
    name="my-new-deployment",
    work_pool_name="work-pool-1",
    job_variables={...},
    build=False,
)

Do NOT call

await Flow.from_source.aio(...)

— that’s the source of the

cls

error. You could technically call

await Flow.from_source.aio(Flow, ...)

to pass the class explicitly, but it’s awkward and unnecessary. Two other important gotchas - Entrypoint format: It must include both the file and the function, like

./src/myproj/flows.py:process_item

. Your example

"path/to/my/flow"

is incomplete and will not register a flow. - Source that works in Kubernetes: If you set

build=False

and rely on your container image to carry your code, make sure: - The image really includes your flow code at the same path as your entrypoint. - Alternatively, use a Git URL for

source

so Prefect can fetch code at runtime (e.g.,

source="<https://github.com/your-org/your-repo>"

with a valid entrypoint path inside the repo). Quick verification steps - List deployments after creation:

prefect deployment ls

- If nothing shows up, surface the exception instead of swallowing it (remove the “Continuing anyway” print) to see the real error. - Ensure the work pool exists:

prefect work-pool ls
  prefect work-pool inspect work-pool-1

- Start a Kubernetes worker for that pool:

prefect worker start --pool work-pool-1

Docs - Deployments: Create deployments - Kubernetes: prefect-kubernetes If you share your exact entrypoint (file path + flow function) and how your image is built, I can sanity-check the

source

and

entrypoint

combo to ensure it’ll resolve correctly in the Kubernetes pod.

✅ thought for 58.7 seconds

You’re still hitting the async wrapper. The tell is this line in your traceback: RuntimeWarning: coroutine 'Flow.afrom_source' was never awaited That means you’re getting the async alias for the classmethod instead of calling the sync classmethod. As a result,

flow_obj

is a coroutine, so

flow_obj.deploy(...)

fails with "'coroutine' object has no attribute 'deploy'". Fix: don’t use .aio (or afrom_source) on Flow.from_source. Call the sync classmethod and the sync deploy. It’s safe to call these inside an async function. Drop-in working example - Make sure your entrypoint includes both the file and the flow function name. - Only use the sync methods.

from pathlib import Path
from prefect.flows import Flow

async def create_deployment_if_needed():
    # 1) Load the flow from your source (sync classmethod)
    flow_obj = Flow.from_source(
        source=str(Path(__file__).resolve().parents[5]),
        # MUST be "path/to/file.py:flow_function_name"
        entrypoint="path/to/my/flow_file.py:process_item",
    )

    # 2) Create/update the deployment (sync instance method)
    deployment_id = flow_obj.deploy(
        name="my-new-deployment",
        work_pool_name="work-pool-1",
        job_variables={
            "image": "my-docker-image:latest",
            "namespace": "my-namespace",
            "image_pull_policy": "Always",
        },
        build=False,   # your image already has your code/deps
        push=False,    # set True if Prefect should build/push
        print_next_steps=False,
    )
    print(f":white_check_mark: Created/updated deployment with ID: {deployment_id}")

If you absolutely must use the async wrappers You can, but you need to pass the class explicitly to the classmethod’s async wrapper, then use the async wrapper for the instance method too:

flow_obj = await Flow.from_source.aio(
    Flow,
    source="<https://github.com/your-org/your-repo>",
    entrypoint="src/myproj/flows.py:process_item",
)
deployment_id = await flow_obj.deploy.aio(
    name="my-new-deployment",
    work_pool_name="work-pool-1",
    job_variables={...},
    build=False,
)

Common gotchas to double-check - Entrypoint format: must be file.py:flow_func_name. If you pass only a path, Prefect won’t find a flow to deploy. - Source path: if you use a local filesystem path, your runtime pods must also have that code (e.g., baked into your image). Otherwise, prefer a Git URL for source. - Don’t swallow exceptions: remove “Continuing anyway” until you see the actual failure. Then verify:

prefect deployment ls
  prefect work-pool inspect work-pool-1

- Make sure you’re importing the right Flow:

from prefect.flows import Flow
  # not the decorator `flow`

Docs for reference - Create deployments: https://docs.prefect.io/v3/how-to-guides/deployments/create-deployments - Kubernetes integration: https://docs.prefect.io/integrations/prefect-kubernetes/ If this still errors, paste your exact

source

and

entrypoint

values and I’ll sanity-check the paths against your image/repo layout.

Thanks for the concrete snippet — that helps. The error you’re seeing means

flow_obj

is a coroutine instead of a Flow, so

flow_obj.deploy(...)

fails. The giveaway is this runtime warning in your earlier trace: RuntimeWarning: coroutine 'Flow.afrom_source' was never awaited What’s happening - In your environment,

Flow.from_source

is resolving to the async path and returning a coroutine. Since you don’t await it,

flow_obj

becomes a coroutine, not a Flow instance. Two solid ways to fix it Option 1 (recommended): Explicitly await Flow.from_source in your async function, keep deploy synchronous

from pathlib import Path
from prefect.flows import Flow

async def create_deployment_if_needed():
    flow_obj = await Flow.from_source(
        source=str(Path(__file__).resolve().parents[5]),
        entrypoint="core/prefect_pipelines/pipelines/chinese_account_registration/igxe_web_registration/flows/igxe_web_process_single_registration.py:process_single_igxe_registration_flow",
    )

    deployment_id = flow_obj.deploy(
        name="igxe-single-registration-k8s",
        work_pool_name="k8s-csgo-stats",
        job_variables={
            # IMPORTANT: use a plain image name, not Slack link markup
            "image": "<http://ghcr.io/stefanignjatovic12/csgo-stats:latest|ghcr.io/stefanignjatovic12/csgo-stats:latest>",
            "namespace": "csgo-stats",
            "image_pull_policy": "Always",
        },
        build=False,
        push=False,
        print_next_steps=False,
    )
    print(f":white_check_mark: Created/updated deployment with ID: {deployment_id}")

Option 2: Use the explicit async wrapper for the classmethod

flow_obj = await Flow.from_source.aio(
    Flow,
    source=str(Path(__file__).resolve().parents[5]),
    entrypoint="core/prefect_pipelines/pipelines/chinese_account_registration/igxe_web_registration/flows/igxe_web_process_single_registration.py:process_single_igxe_registration_flow",
)
deployment_id = flow_obj.deploy(
    name="igxe-single-registration-k8s",
    work_pool_name="k8s-csgo-stats",
    job_variables={ ... },
    build=False,
    push=False,
)

Key things to double-check - Entrypoint format: Must be file.py:flow_function_name. Yours looks correct. - Image string: In your snippet you pasted a Slack-formatted link as the image value: "image": "ghcr.io/stefanignjatovic12/csgo-stats:latest" Kubernetes expects a plain image name like: "image": "ghcr.io/stefanignjatovic12/csgo-stats:latest" - Source path vs runtime: With build=False, your container image must contain the flow code at the same path the entrypoint expects. If that’s not true, prefer a Git URL for source (e.g., "https://github.com/your-org/your-repo") and make sure the repo has the entrypoint path. Sanity checks - After it runs, verify the deployment exists:

prefect deployment ls

- Verify your work pool:

prefect work-pool inspect k8s-csgo-stats

- Start a worker if it’s not already running:

prefect worker start --pool k8s-csgo-stats

Reference: Deploy via Python If you still see a coroutine after awaiting

Flow.from_source

, share your Prefect version (

prefect version

) and I’ll adjust the guidance.