apiVersion: batch/v1
kind: Job
spec:
  template:
    spec:
      nodeSelector:
        nodepool: high-mem
      # optionally also:
      # tolerations:
      # affinity:

--job-template /path/to/job_template.yaml

agent:
  extraVolumes:
    - name: job-template
      configMap:
        name: prefect-job-template
  extraVolumeMounts:
    - name: job-template
      mountPath: /opt/prefect/job-template
      readOnly: true
  extraArgs:
    - --job-template
    - /opt/prefect/job-template/job_template.yaml

run_config

from prefect import Flow, task
from prefect.run_configs import KubernetesRun

job_template = {
  "apiVersion": "batch/v1",
  "kind": "Job",
  "spec": {
    "template": {
      "spec": {
        "nodeSelector": {"nodepool": "high-mem"}
      }
    }
  }
}

@task
def do_work():
    pass

with Flow("my-flow", run_config=KubernetesRun(job_template=job_template)) as flow:
    do_work()

KubernetesRun(job_template_path="job_template.yaml")

<http://3928...|3928...>

nodeSelector

"<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"

<http://kubernetes.io/arch|kubernetes.io/arch>

<http://392865356492.dkr.ecr.us-east-1.amazonaws.com/eashared-prefectv1-dev:pp-9775c87|392865356492.dkr.ecr.us-east-1.amazonaws.com/eashared-prefectv1-dev:pp-9775c87>

{ "<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64" }

{
  "apiVersion": "batch/v1",
  "kind": "Job",
  "metadata": {},
  "spec": {
    "template": {
      "metadata": {},
      "spec": {
        "restartPolicy": "Never",
        "nodeSelector": {
          "<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"
        }
      }
    }
  }
}

run_config=KubernetesRun(job_template_path="/opt/prefect/job-template/job_template.yaml")

apiVersion: batch/v1
kind: Job
spec:
  template:
    spec:
      restartPolicy: Never
      nodeSelector:
        <http://kubernetes.io/arch|kubernetes.io/arch>: amd64

apiVersion: batch/v1
kind: Job
spec:
  template:
    spec:
      restartPolicy: Never
      nodeSelector:
        <http://kubernetes.io/arch|kubernetes.io/arch>: amd64

apiVersion: batch/v1
kind: Job
spec:
  template:
    spec:
      restartPolicy: Never
      nodeSelector:
        <http://kubernetes.io/arch|kubernetes.io/arch>: amd64
      containers:
        - name: flow
          resources:
            requests:
              cpu: "250m"
              memory: "512Mi"
            limits:
              memory: "1Gi"
          # add env, volumeMounts, etc. here if needed

"<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"

<http://kubernetes.io/arch|kubernetes.io/arch>

job_template = {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "spec": {
        "template": {
            "spec": {
                "nodeSelector": {
                    "<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"
                },
                "restartPolicy": "Never"
            }
        }
    }
}

spec.template.spec.containers

containers: []

spec:
  template:
    spec:
      nodeSelector:
        <http://kubernetes.io/arch|kubernetes.io/arch>: amd64
      containers:
        - name: flow
          resources:
            requests:
              cpu: "250m"
              memory: "512Mi"
            limits:
              memory: "1Gi"

apiVersion: batch/v1
    kind: Job
    spec:
      template:
        spec:
          restartPolicy: Never
          nodeSelector:
            <http://kubernetes.io/arch|kubernetes.io/arch>: amd64

run_config = KubernetesRun(
        image=image,
        labels=list(labels) if labels else [],
        memory_request="512Mi",
        memory_limit="1Gi",
        cpu_request="250m",
        job_template_path="/opt/prefect/job-template/job_template.yaml",
    )

PREFECT__LOGGING__LEVEL=DEBUG

containers

apiVersion: batch/v1
    kind: Job
    spec:
      template:
        spec:
          restartPolicy: Never
          nodeSelector:
            <http://kubernetes.io/arch|kubernetes.io/arch>: amd64

job_template = {
      "apiVersion": "batch/v1",
      "kind": "Job",
      "spec": {
        "template": {
          "spec": {
            "restartPolicy": "Never",
            "nodeSelector": {"<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"}
          }
        }
      }
    }

kubectl -n <ns> get pods | grep agent

kubectl get pods -A | grep agent

kubectl -n <ns> logs -f deployment/<agent-deployment-name>

kubectl -n <ns> logs -f <agent-pod-name>

kubectl -n <ns> get deploy -l <http://app.kubernetes.io/instance=<release-name|app.kubernetes.io/instance=<release-name>>

kubectl -n <ns> get deploy <agent-deployment-name> -o yaml

--job-template /path/to/job_template.yaml

PREFECT__LOGGING__LEVEL

PREFECT__CLOUD__AGENT__KUBERNETES__JOB_TEMPLATE_PATH

agent:
    extraEnv:
      - name: PREFECT__LOGGING__LEVEL
        value: DEBUG
    extraArgs:
      - --log-level
      - DEBUG

kubectl -n <ns> get deploy <agent-deployment-name> -o yaml

--job-template /path/to/job_template.yaml

PREFECT__CLOUD__AGENT__KUBERNETES__JOB_TEMPLATE_PATH

PREFECT__AGENT__KUBERNETES__JOB_TEMPLATE_PATH

apiVersion: batch/v1
  kind: Job
  spec:
    template:
      spec:
        restartPolicy: Never
        nodeSelector:
          <http://kubernetes.io/arch|kubernetes.io/arch>: amd64

job_template = {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "spec": {
      "template": {
        "spec": {
          "restartPolicy": "Never",
          "nodeSelector": {
            "<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"
          },
          "containers": [
            {"name": "flow"}
          ]
        }
      }
    }
  }

kubectl -n <ns> logs -f deployment/<agent-deployment-name>

from prefect import Flow
from prefect.storage import Docker

flow = Flow("my-flow")
flow.storage = Docker(
  registry_url="<http://392865356492.dkr.ecr.us-east-1.amazonaws.com|392865356492.dkr.ecr.us-east-1.amazonaws.com>",
  image_name="eashared-prefectv1-dev",
  image_tag="pp-9775c87",
  python_dependencies=["prefect==1.4.1"]  # and your deps
)
# flow.register(project_name="...")  # when you register

apt-get update && apt-get install -y git && rm -rf /var/lib/apt/lists/*

apk add --no-cache git openssh

from prefect.storage import Git  # or GitHub
flow.storage = Git(
  repo="<https://github.com/owner/repo.git>",
  flow_path="flows/my_flow.py",
  branch="main"  # or tag/commit
  # If private, put the PAT in the URL or use env/SSH key
)

GIT_SSH_COMMAND=ssh -i /path/key -o StrictHostKeyChecking=no

# Find the pod for this run
kubectl -n <ns> get pods -l "<http://prefect.io/flow-run-id=62e99500-5f8b-48f4-bc73-7a41b6485fd3|prefect.io/flow-run-id=62e99500-5f8b-48f4-bc73-7a41b6485fd3>" -o name
# Then:
kubectl -n <ns> logs <pod-name>

kubectl -n ea-prefect-dev get svc,pods,endpoints | grep -E 'prefectdev-graphql|4201'

from prefect.storage import GitHub

storage = GitHub(
    repo="enverus-ea/ea.data.mfg.prefectV1",  # owner/repo, no .git, no https
    path=flow_file,                           # e.g., "flows/my_flow.py"
    ref=git_ref,                              # branch/tag/commit
    access_token_secret="GITHUB_ACCESS_TOKEN" # name of the secret, not the value
)
flow.storage = storage

run_config = KubernetesRun(
      image=image,
      env={"PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN": os.getenv("GITHUB_ACCESS_TOKEN")}
    )

spec:
      template:
        spec:
          containers:
            - name: flow
              env:
                - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
                  valueFrom:
                    secretKeyRef:
                      name: github-token
                      key: token

kubectl -n <ns> logs <flow-pod>
  kubectl -n <ns> exec -it <flow-pod> -- env | grep GITHUB

Secret("GITHUB_ACCESS_TOKEN")

GITHUB_ACCESS_TOKEN

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

agent:
  env:
    - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
      valueFrom:
        secretKeyRef:
          name: github-token
          key: token
  job:
    env:
      - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
        valueFrom:
          secretKeyRef:
            name: github-token
            key: token

value: "..."

valueFrom

GITHUB_ACCESS_TOKEN

Secret(...)

os.environ["GITHUB_ACCESS_TOKEN"]

Secret("GITHUB_ACCESS_TOKEN")

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

job:

agent.job

agent.job

agent:
  env:
    - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
      valueFrom:
        secretKeyRef:
          name: github-token
          key: token

job:
  env:
    - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
      valueFrom:
        secretKeyRef:
          name: github-token
          key: token

kubectl get pod <flow-run-pod> -o jsonpath='{.spec.containers[0].env}'

kubectl exec <flow-run-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

GITHUB_ACCESS_TOKEN

agent.env

agent.job.env

- name: PREFECT__CLOUD__USE_LOCAL_SECRETS
  value: "true"

GITHUB_ACCESS_TOKEN

kubectl get pod <flow-run-pod> -o jsonpath='{.spec.containers[0].env}'
# or
kubectl exec <flow-run-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

github-credentials

github-access-token

jobTemplateFilePath: ""

Secret("GITHUB_ACCESS_TOKEN")

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

Secret(...).get()

agent:
  env:
    - name: PREFECT__CLOUD__USE_LOCAL_SECRETS
      value: "true"
    - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
      valueFrom:
        secretKeyRef:
          name: github-credentials
          key: github-access-token
  job:
    env:
      - name: PREFECT__CLOUD__USE_LOCAL_SECRETS
        value: "true"
      - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
        valueFrom:
          secretKeyRef:
            name: github-credentials
            key: github-access-token

Secret("GITHUB_ACCESS_TOKEN")

access_token_secret="GITHUB_ACCESS_TOKEN"

Secret("GITHUB_ACCESS_TOKEN")

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

agent.env

agent.job.env

kubectl get pod <flow-run-pod> -o jsonpath='{.spec.containers[0].env}'
kubectl exec <flow-run-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

kubectl get secret github-credentials -o yaml

GITHUB_ACCESS_TOKEN

# values.yaml (flow-run pod)
agent:
  job:
    env:
      - name: GITHUB_ACCESS_TOKEN
        valueFrom:
          secretKeyRef:
            name: github-credentials
            key: github-access-token

# flow code
import os
from prefect.storage import GitHub

storage = GitHub(
    repo=git_repo,
    path=flow_file,
    ref=git_ref,
    access_token=os.environ["GITHUB_ACCESS_TOKEN"],
)

Secret(...)

printenv | grep GITHUB

KubernetesRun

jobs:
  register:
    runs-on: ubuntu-latest
    env:
      GITHUB_ACCESS_TOKEN: ${{ secrets.GH_PAT }}  # or your secret name
    steps:
      - uses: actions/checkout@v4
      - run: python utilities/register-flow.py register-flow

export GITHUB_ACCESS_TOKEN=ghp_...
python utilities/register-flow.py register-flow

from prefect.storage import GitHub

storage = GitHub(
    repo=git_repo,
    path=flow_file,
    ref=git_ref,
    access_token_secret="GITHUB_ACCESS_TOKEN",
)

agent:
  job:
    env:
      - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
        valueFrom:
          secretKeyRef:
            name: github-credentials
            key: github-access-token

kubectl exec <flow-run-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

python -c "import os; print('set' if 'GITHUB_ACCESS_TOKEN' in os.environ else 'missing')"

GITHUB_TOKEN

env:
  GITHUB_ACCESS_TOKEN: ${{ secrets.GITHUB_TOKEN }}

access_token=os.environ["GITHUB_ACCESS_TOKEN"]

access_token_secret="GITHUB_ACCESS_TOKEN"

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

agent.job.env

# GitHub Actions example
env:
  GITHUB_ACCESS_TOKEN: ${{ secrets.GH_PAT }}

from prefect.storage import GitHub

storage = GitHub(
    repo=git_repo,
    path=flow_file,
    ref=git_ref,
    access_token_secret="GITHUB_ACCESS_TOKEN",  # not os.environ[…]
)

kubectl exec <flow-run-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

kubectl get secret github-credentials -n <ns> -o yaml

- name: PREFECT__CLOUD__USE_LOCAL_SECRETS
  value: "true"

printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

GitHub

access_token_secret

access_token

from prefect.storage import GitHub

storage = GitHub(
    repo=git_repo,
    path=flow_file,
    ref=git_ref,
    access_token_secret="GITHUB_ACCESS_TOKEN",
)

Secret("GITHUB_ACCESS_TOKEN")

agent:
  env:
    - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
      valueFrom:
        secretKeyRef:
          name: github-credentials
          key: github-access-token
  job:
    env:
      - name: PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
        valueFrom:
          secretKeyRef:
            name: github-credentials
            key: github-access-token

kubectl exec <flow-run-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

containers

env

agent.job.env

job_template

env: []

job_template = {
  "apiVersion": "batch/v1",
  "kind": "Job",
  "spec": {
    "template": {
      "spec": {
        "restartPolicy": "Never",
        "nodeSelector": {"<http://kubernetes.io/arch|kubernetes.io/arch>": "amd64"},
        "containers": [{
          "name": "flow",
          "env": [
            {
              "name": "PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN",
              "valueFrom": {
                "secretKeyRef": {
                  "name": "github-credentials",
                  "key": "github-access-token",
                }
              }
            }
          ]
        }]
      }
    }
  }
}

env

agent.job.env

kubectl get pod <flow-pod> -o jsonpath='{.spec.containers[0].env}'

kubectl exec <flow-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

jobTemplateFilePath

job_template

containers/env

KubernetesRun(job_template=...)

agent.job.env

job_template

import tasks

from prefect.storage import Git

storage = Git(
    repo="<https://github.com/your-org/your-repo.git>",
    flow_path=flow_file,          # path to the flow file within the repo
    ref=git_ref,                  # branch/tag/sha
    git_token_secret="GITHUB_ACCESS_TOKEN",  # same Prefect secret name
)

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

git_token_secret

pip install .

from prefect.storage import Module

storage = Module("your_package.flows.my_flow")

job_template = {
  "apiVersion": "batch/v1",
  "kind": "Job",
  "spec": {
    "template": {
      "spec": {
        "restartPolicy": "Never",
        "volumes": [{"name": "repo", "emptyDir": {}}],
        "initContainers": [{
          "name": "git-clone",
          "image": "alpine/git:2.45.2",
          "env": [{
            "name": "GITHUB_TOKEN",
            "valueFrom": {"secretKeyRef": {"name": "github-credentials", "key": "github-access-token"}}
          }],
          "command": ["sh","-c"],
          "args": ["git clone --branch ${GIT_REF:-main} https://$GITHUB_TOKEN@github.com/your-org/your-repo.git /repo"],
          "volumeMounts": [{"name": "repo", "mountPath": "/repo"}]
        }],
        "containers": [{
          "name": "flow",
          "env": [{"name": "PYTHONPATH", "value": "/repo:${PYTHONPATH}"}],
          "volumeMounts": [{"name": "repo", "mountPath": "/repo"}]
        }]
      }
    }
  }
}

import tasks

access_token_secret="..."

git clone

git_token_secret="..."

from prefect.storage import Git

storage = Git(
    repo="<https://github.com/your-org/your-repo.git>",
    flow_path="path/to/flow.py",
    ref="main",
    git_token_secret="GITHUB_ACCESS_TOKEN",  # resolves PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
)

Git

ref

from prefect.storage import Git

storage = Git(
    repo="<https://github.com/your-org/your-repo.git>",
    flow_path=flow_file,              # path to the flow file in the repo
    branch="main",                    # not 'ref'
    git_token_secret="GITHUB_ACCESS_TOKEN",  # resolves PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
)

branch_name

branch

branch

storage = Git(..., branch_name="main", git_token_secret="GITHUB_ACCESS_TOKEN")

Git

repo=...

NotGitRepository()

from prefect.storage import Git

storage = Git(
    repo="<https://github.com/your-org/your-repo.git>",
    flow_path="path/to/flow.py",     # required; path to your flow in the repo
    branch="main",                   # or your branch/tag
    git_token_secret="GITHUB_ACCESS_TOKEN",
)

branch

branch_name="main"

RUN apt-get update && apt-get install -y git && rm -rf /var/lib/apt/lists/*
RUN pip install gitpython

git --version
python -c "import git, sys; print('gitpython ok', git.__version__)"

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

kubectl exec <flow-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

flow_path

repo="<http://github.com/your-org/your-repo|github.com/your-org/your-repo>"   # looks like a local path to GitPython

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

Git(...)

git --version

<https://github.com/your-org/your-repo.git>

branch

branch_name

git_token_secret

git_token_secret_name

from prefect.storage import Git

storage = Git(
    repo="<https://github.com/your-org/your-repo.git>",
    flow_path=flow_file,                 # path to the flow file within the repo
    branch=git_ref,                      # branch/tag; SHA may not be supported in all versions
    git_token_secret="GITHUB_ACCESS_TOKEN",
)

RUN apt-get update && apt-get install -y git && rm -rf /var/lib/apt/lists/*
RUN pip install gitpython

git --version
python -c "import git; print('gitpython ok')"

flow

env: []

env

kubectl exec <flow-pod> -- printenv | grep PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

kubectl exec <flow-pod> -- sh -c 'git --version && python -c "import git; print(1)"'

repo=

from tasks import …

FROM prefecthq/prefect:1.4-python3.8  # or your base
WORKDIR /app
COPY . /app
RUN pip install -U pip && pip install .
# or: RUN pip install -e .
# If you didn’t package it, at least expose it
ENV PYTHONPATH=/app:${PYTHONPATH}

job_template = {
  "apiVersion": "batch/v1",
  "kind": "Job",
  "spec": {
    "template": {
      "spec": {
        "restartPolicy": "Never",
        "volumes": [{"name": "repo", "emptyDir": {}}],
        "initContainers": [{
          "name": "git-clone",
          "image": "alpine/git:2.45.2",
          "env": [
            {
              "name": "GITHUB_TOKEN",
              "valueFrom": {
                "secretKeyRef": {"name": "github-credentials", "key": "github-access-token"}
              }
            },
            {"name": "GIT_REF", "value": "main"}  # or pass your ref via env
          ],
          "command": ["sh","-c"],
          "args": [
            "git clone --depth 1 --branch \"$GIT_REF\" "
            "https://$GITHUB_TOKEN@github.com/your-org/your-repo.git /repo"
          ],
          "volumeMounts": [{"name": "repo", "mountPath": "/repo"}]
        }],
        "containers": [{
          "name": "flow",
          "env": [{"name": "PYTHONPATH", "value": "/repo:${PYTHONPATH}"}],
          "volumeMounts": [{"name": "repo", "mountPath": "/repo"}]
        }]
      }
    }
  }
}

PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN

import os, sys, subprocess

def _ensure_repo():
    repo_dir = "/tmp/repo"
    if not os.path.exists(os.path.join(repo_dir, ".git")):
        from prefect import Secret
        token = Secret("GITHUB_ACCESS_TOKEN").get()
        ref = os.getenv("GIT_REF", "main")
        url = f"https://{token}:x-oauth-basic@github.com/your-org/your-repo.git"
        subprocess.check_call(["git", "clone", "--depth", "1", "--branch", ref, url, repo_dir])
    if repo_dir not in sys.path:
        sys.path.insert(0, repo_dir)

_ensure_repo()

from tasks import something

.git

from prefect import flow

    @flow
    def my_flow(): ...

    if __name__ == "__main__":
        (
            my_flow
            .from_source(
                source="<https://github.com/OWNER/REPO>",
                entrypoint="path/to/flow.py:my_flow"
            )
            .deploy(name="prod", work_pool_name="your-pool")
        )

.git

git

.git

.git

.dockerignore

RUN git clone --branch <branch> <url> /opt/flows

/opt/flows/.git

.git

flow.from_source(...)

git

.git

.git

.git

prefect.storage.Git

https://...git

git@...

.git

prefect.storage.Docker

.git

.git

.git

.dockerignore

RUN git clone --branch main --depth 1 <https://github.com/OWNER/REPO.git> /opt/flow
    WORKDIR /opt/flow

Git(repo="/opt/flow", flow_path="path/to/flow.py", branch="main")

pwd && ls -a
  # Expect to see .git in the directory Prefect is using

.git

repo

flow.storage = Git(...)

api_token

api_token=

access_token_secret

from prefect.storage import GitHub

    flow.storage = GitHub(
        repo="OWNER/REPO",
        path="path/to/flow.py",
        ref="main",
        access_token_secret="GITHUB_ACCESS_TOKEN"  # name of a Prefect Secret
    )

git_token_secret_name

from prefect.storage import Git

    flow.storage = Git(
        repo="<https://github.com/OWNER/REPO.git>",
        flow_path="path/to/flow.py",
        ref="main",
        git_token_secret_name="GIT_TOKEN"  # name of a Prefect Secret
        # or: use_ssh=True and mount an SSH key instead
    )

api_token

Client(api_token=...)

prefect backend server

PREFECT__CLOUD__*

api_token

flow.storage = ...

cloud = prefect.config.cloud

api_token

api_token

from prefect.client import Client
import prefect

class GraphQLCloudAPI:
    def __init__(self) -> None:
        # Option A: rely on PREFECT__BACKEND=server and PREFECT__SERVER__ENDPOINT env vars
        self.client = Client()

        # Option B: set explicitly if you prefer
        # self.client = Client(api_server=prefect.config.server.endpoint)

Client()

api_key

api_token

PREFECT__BACKEND=server

PREFECT__SERVER__ENDPOINT=http://<host>:4200/graphql

prefect.config.cloud.*

prefect.config.backend

backend == "cloud"

prefect.__version__

from prefect.client import Client

self.client = Client(api_server="http://<your-server-host>:4200/graphql")

from prefect.client import Client
import prefect

self.client = Client(api_server=prefect.config.server.endpoint)

PREFECT__BACKEND=server
PREFECT__SERVER__ENDPOINT=http://<your-server-host>:4200/graphql

self.client = Client()

flow.run()

flow_run_id

prefect execute flow-run

--flow-run-id

task_id

flow_run_id

prefect execute flow-run --flow-run-id <UUID>

import prefect
  logger = prefect.context.logger
  <http://logger.info|logger.info>(f"flow_run_id={prefect.context.get('flow_run_id')}")

flow.run()

task_id: ""

flow.run()

ref

query($fr: uuid!) {
    flow_run(where: {id: {_eq: $fr}}) {
      flow {
        id
        name
        tasks { id name slug }
      }
    }
  }

# flows/my_flow.py
<code>

# storage.py
<code>

# register.py
<code>

task_id: ""

flow.run()

Task(...)

python -c "import prefect; print(prefect.__version__)"

task_id

from prefect import task, Flow

  @task(name="extract", slug="extract")
  def extract(): ...

  @task(name="transform", slug="transform")
  def transform(x): ...

  with Flow("my-flow") as flow:
      t = transform(extract())

from my_lib.tasks import shared_task

  step1 = shared_task.copy(name="shared_step1", slug="shared_step1")
  step2 = shared_task.copy(name="shared_step2", slug="shared_step2")

import prefect
  from prefect.utilities.logging import get_logger
  logger = get_logger()

  @task
  def debug():
      ctx = prefect.context
      <http://logger.info|logger.info>(f"flow_run_id={ctx.get('flow_run_id')}, task_slug={ctx.get('task_slug')}, task_full_name={ctx.get('task_full_name')}")

python -c "import prefect; print(prefect.__version__)"

RUN pip install "prefect==1.4.1"

python -c "import prefect; print(prefect.__version__)"  # should print 1.4.1

prefect backend server
  # Or ensure PREFECT__BACKEND=server and PREFECT__SERVER__ENDPOINT is set appropriately

ref

ref

.copy()

# apollo
    FROM prefecthq/apollo:2022.09.23
    # graphql
    FROM prefecthq/graphql:2022.09.23
    # ui
    FROM prefecthq/ui:2022.09.23
    # hasura (use the tag the 2022.09.23 chart expects)
    FROM hasura/graphql-engine:v1.3.3

RUN pip install "prefect==1.4.1"

python -c "import prefect; print(prefect.__version__)"  # should be 1.4.1

PREFECT__BACKEND=server

PREFECT__SERVER__ENDPOINT=http://<server>:4200/graphql

Client(api_server=...)

prefecthq/server:core-:2022.09.23

prefecthq/server:2022.09.23

prefecthq/apollo:2022.09.23

prefecthq/ui:2022.09.23

hasura/graphql-engine:v1.3.3

pip install "prefect==1.4.1"

core-1.4.1

FROM prefecthq/server:2022.09.23      # for the GraphQL service
  FROM prefecthq/apollo:2022.09.23
  FROM prefecthq/ui:2022.09.23
  FROM hasura/graphql-engine:v1.3.3     # Hasura

serverVersionTag: "2022.09.23"
  prefectVersionTag: "2022.09.23"
  uiVersionTag: "2022.09.23"

prefect==1.4.1

# Server stack (all must match)
FROM prefecthq/server:2022.09.23      # GraphQL service
FROM prefecthq/apollo:2022.09.23
FROM prefecthq/ui:2022.09.23
FROM hasura/graphql-engine:v1.3.3.cli-migrations-v2

# Agents / flow runtime
FROM prefecthq/prefect:1.4.1

serverVersionTag: "core-1.4.1"
prefectVersionTag: "core-1.4.1"
uiVersionTag: "core-1.4.1"
hasura:
  image:
    tag: "v2.0.1.cli-migrations-v2"

serverVersionTag: "core-1.4.1"
prefectVersionTag: "core-1.4.1"
uiVersionTag: "core-1.4.1"

hasura:
  image:
    repository: hasura/graphql-engine
    tag: "v2.0.1.cli-migrations-v2"

server:
  image:
    repository: prefecthq/server
    tag: core-1.4.1

apollo:
  image:
    repository: prefecthq/apollo
    tag: core-1.4.1

ui:
  image:
    repository: prefecthq/ui
    tag: core-1.4.1

hasura:
  image:
    repository: hasura/graphql-engine
    tag: v2.0.1.cli-migrations-v2

step1 = shared_task.copy(name="shared_step1", slug="shared_step1")
  step2 = shared_task.copy(name="shared_step2", slug="shared_step2")

from my_pkg.flow import flow
  print(sorted(t.slug for t in flow.tasks))

flow.run()

import prefect
  from prefect import task
  @task(name="debug_ctx", slug="debug_ctx")
  def debug_ctx():
      ctx = prefect.context
      print("flow_run_id:", ctx.get("flow_run_id"))
      print("task_slug:", ctx.get("task_slug"))

task_id: ""

from prefect import task, Flow

@task(name="extract", slug="extract")
def extract(): ...

@task(name="transform", slug="transform")
def transform(x): ...

with Flow("my-flow") as flow:
    t1 = extract()
    t2 = transform(t1)

with Flow(...)

from my_tasks import extract, transform, load
with Flow("etl") as flow:
    data = extract()
    clean = transform(data)
    load(clean)

flow.add_task(extract)
flow.add_task(transform)
flow.set_dependencies(task=transform, upstream_tasks=[extract])

print("tasks:", [t.slug for t in flow.tasks])
print("task count:", len(flow.tasks))

prefect backend server

prefect create project "my-project"

flow.register(project_name="my-project")

from my_lib.tasks import shared_task
a = shared_task.copy(name="step_a", slug="step_a")
b = shared_task.copy(name="step_b", slug="step_b")

from my_pkg.flow import flow
print(sorted(t.slug for t in flow.tasks))