Thanks for the detail — that single log line is a strong clue. What that message means - “Importing flow code from module path app.core.tasks.ingest_asset_use_case” is emitted when the worker starts executing the flow run and tries to import your flow module (entrypoint is a module path, not a file path). - If nothing else appears and the run stays in Pending, the import is likely hanging (the flow never gets to “Running”). Most common causes of import-time hangs - Module-level I/O or heavy work: network calls (API requests, cloud SDK auth), DB connections, reading large files, etc., executed at import time. - Long/blocked cloud SDK credential lookups (boto3, GCP, etc.) during import. - Circular imports between your modules. - Environment access at import time that blocks (e.g., DNS/connectivity issues). - Very slow imports or side-effects from libraries initialized globally. Fast diagnostics to pinpoint the hang 1) Try importing the module inside the worker’s environment - Open a shell in the worker container/VM and run:

python - <<'PY'
import time, importlib
t = time.time()
print("Starting import…")
mod = importlib.import_module("app.core.tasks.ingest_asset_use_case")
print("Import finished in", round(time.time()-t, 2), "s")
# If that works, try Prefect’s loader too:
from prefect.flows import load_flow_from_entrypoint
flow = load_flow_from_entrypoint("app.core.tasks.ingest_asset_use_case", use_placeholder_flow=False)
print("Flow loaded:", flow.name)
PY

- If this hangs, the issue is inside module import itself (not Prefect). 2) Increase logging on the worker - Set

PREFECT_LOGGING_LEVEL=DEBUG

for the worker and re-run. You should see where it stops after that “Importing…” message. 3) Use Python faulthandler to grab a traceback when it’s stuck - Run the worker with

PYTHONFAULTHANDLER=1

. When it appears stuck, send SIGUSR1 to the worker process to dump stack traces: - Linux host:

kill -USR1 <worker_pid>

- Docker:

docker kill --signal=USR1 <container_id>

- The traceback will show the exact line that’s blocking during import. 4) Add temporary “bracket” logs in your module - At the very top of `app/core/tasks/ingest_asset_use_case.py`:

import logging, time
log = logging.getLogger(__name__)
log.warning("ingest_asset_use_case: import started")
# … your imports …
log.warning("ingest_asset_use_case: imports complete")
# … any module-level setup …
log.warning("ingest_asset_use_case: module-level init complete")

- Redeploy/restart worker and watch which message is last. How to fix (best practices) - Avoid module-level work. Move it into tasks or into the flow function so it happens during execution, not import:

from prefect import flow, task

@task
def get_config():
    import requests
    return requests.get("<https://api.example.com/config>", timeout=10).json()

@flow
def ingest_asset_use_case():
    cfg = get_config()
    # rest of your logic

- If you must initialize clients at module load, ensure they don’t connect yet and use short timeouts: - Databases: create engine with timeouts and connect inside tasks:

from sqlalchemy import create_engine
engine = create_engine(
    "postgresql://…",
    pool_pre_ping=True,
    connect_args={"connect_timeout": 5},
)

- HTTP/SDKs: never call remote endpoints at import; set credentials from env and defer first network call to task/flow. - Eliminate circular imports by moving shared code to a third module used by both sides. - If the worker runs in a restricted network, verify it has egress/permissions for anything your module touches. Many SDKs try metadata/IMDS calls on import which can stall in locked-down networks. - Sanity‑check that your module defines exactly one flow if you’re using a module path entrypoint; if there are multiple flows, prefer targeting the specific function with a file path entrypoint like

path/to/file.py:flow_func

(though note the module still imports). A couple of quick questions to tailor next steps

- Are you on Prefect 3.x, and what worker type/infrastructure are you using (process, Docker, K8s, ECS)? - Does the module

app.core.tasks.ingest_asset_use_case

do any DB connects, API calls, or client creation at the top level? - Can you run the import test above inside the worker environment and share whether it hangs and where the faulthandler traceback points? If you can share the top ~30 lines of that module (with secrets removed), I can point out the likely culprit and suggest a minimal refactor.

Got it — thanks. Since you’re on Prefect 3.6.6 with a process worker in Kubernetes and all flows live in a single app/core/tasks.py, the “Importing flow code from module path …” log followed by a Pending hang almost always means the module import is blocking. Why this likely affects you - With all flows in one file, importing any one flow imports the entire module. Any module-level work (DB connects, cloud SDK initialization, network calls, heavy imports, reading files) will run before the flow can start, so the run can sit in Pending. - In K8s, some SDKs (AWS/GCP/Azure) try metadata/IMDS calls or credential discovery at import time that can stall if the pod can’t reach metadata services. Quick isolations 1) Try a tiny flow in its own minimal module - Create app/core/hello.py:

from prefect import flow

@flow
def hello():
    return "ok"

- Deploy it with a file entrypoint:

prefect deploy app/core/hello.py:hello -n hello

- If this runs, the worker is fine and the issue is specific to app/core/tasks.py side-effects. 2) Import test inside the worker pod - Exec into the worker pod and run:

python - <<'PY'
import time, importlib
t = time.time()
print("Starting import…")
mod = importlib.import_module("app.core.tasks")
print("Import finished in", round(time.time()-t, 2), "s")
from prefect.flows import load_flow_from_entrypoint
flow = load_flow_from_entrypoint("app.core.tasks.ingest_asset_use_case", use_placeholder_flow=False)
print("Flow loaded:", flow.name)
PY

- If this hangs, it’s definitely import-time work in tasks.py (or something it imports). 3) Add “bracket” logs at module top

import logging
log = logging.getLogger(__name__)
log.warning("tasks.py: import start")
# … top-level imports …
log.warning("tasks.py: imports finished")
# … any module-level initialization …
log.warning("tasks.py: module-level init finished")

Re-run and see which line is last in your worker logs. Remediations/best practices - Move side-effects out of module scope: - Create DB engines/clients without connecting at import time; connect inside tasks/flow and set short connect timeouts. - Defer network calls/config fetches to tasks or the flow body. - Avoid reading large files at import; do it in tasks. - Split flows into smaller modules so importing one flow doesn’t import everything. - Prefer file entrypoints to be explicit:

prefect deploy app/core/tasks.py:ingest_asset_use_case -n ingest-asset

(Note: this still executes module-level code; it just makes the target explicit.) - If you use cloud SDKs, ensure credentials are provided via env and avoid metadata discovery that can stall: - AWS: set

AWS_EC2_METADATA_DISABLED=true

if you don’t want IMDS, and ensure `AWS_REGION`/`AWS_DEFAULT_REGION` is set. - GCP/Azure: ensure credentials are in env/secret blocks so the library doesn’t try metadata servers during import. - Watch for circular imports if tasks.py imports modules that import back into tasks.py. A couple of targeted questions to zero in: - Does app/core/tasks.py (or imports it pulls in) create DB engines, cloud clients (boto3, google-cloud-*, azure), or make any API/file system calls at the top level? - Can you run the import test in the worker pod and share where it stops? - Are your deployments using the module path

app.core.tasks.ingest_asset_use_case

or the file path

app/core/tasks.py:ingest_asset_use_case

? If it’s the former, can you try the latter to confirm behavior is the same? If you can paste the top ~50 lines of tasks.py (with secrets removed), I can call out the specific lines likely causing the block and suggest a minimal refactor.