flow = Flow("my-flow")
flow.storage = GitLab(repo="my/repo", path="/flow.py", ref="MISOCU-3337")

flow.py

/flow.py

ModuleNotFoundError

prefect ModuleNotFoundError: No module named 'app'

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

from app.job.experiment import ExperimentJob


@task
def process():
    logger = prefect.context.get("logger")
    experiment_job = ExperimentJob(logger)
    experiment_job.update_experiment_user_count()


with Flow(
        "update_experiment_user_count",
        run_config=KubernetesRun(
            image_pull_policy="Always",
            labels=["k8s"],
            image="<http://xxx.dkr.ecr.us-west-2.amazonaws.com/prefect-test|xxx.dkr.ecr.us-west-2.amazonaws.com/prefect-test>",
        ),
        storage=GitLab(
            access_token_secret="gitlab-access-token",
            repo="my/repo",
            ref="MISOCU-3337",
            path="flow.py"
        ),
) as flow:
    process()

flow.register(project_name="test_project", labels=["k8s"])

setup.py

from setuptools import setup, find_packages

with open('requirements.txt') as f:
    requirements = f.read().splitlines()

setup(
    name="flow_utilities",
    version="0.1",
    packages=find_packages(),
    install_requires=requirements,
)

flow_test.py

import platform
import pandas as pd
import prefect
from prefect import Flow, Parameter, task
from prefect.client.secrets import Secret
from prefect.storage import GitHub
from prefect.run_configs import KubernetesRun
import subprocess

PREFECT_PROJECT_NAME = "testing"
FLOW_NAME = "flow_test"
AGENT_LABEL = "etl"

STORAGE = GitHub(
    repo="mateo2181/prefect-flows-test",
    path=f"flows/{FLOW_NAME}.py",
    access_token_secret="GITHUB_ACCESS_TOKEN"
)

RUN_CONFIG = KubernetesRun(labels=[AGENT_LABEL])

@task
def extract_and_load(dataset: str) -> None:
    logger = prefect.context.get("logger")
    file = f"<gs://football_transfers/transfers/{dataset}>"
    df = pd.read_csv(file)
    <http://logger.info|logger.info>("Dataset %s with %d rows loaded to DB", dataset, len(df))

@task(log_stdout=True)
def hello_world():
    print(f"Hello from {FLOW_NAME} v2!")
    print(f"Running this task with Prefect: {prefect.__version__} and Python {platform.python_version()}")


with Flow(FLOW_NAME, storage=STORAGE, run_config=RUN_CONFIG,) as flow:
    # user_input = Parameter("user_input", default="Marvin")
    hw = hello_world()
    extract_and_load("1999_dutch_eredivisie.csv")