from prefect import Flow
from prefect.tasks.shell import ShellTask
from prefect.client import Secret

environment = {}
secret_key = Secret("SECRET_KEY")
environment['SECRET_KEY'] = secret_key.get()

with Flow(name, schedule=schedule) as flow:
  task(command='./do-the-thing', env=environment)
  flow.register(project_name=project_name)

prefect.client.Secret

# register-tasks.py
from prefect import Flow
from prefect.tasks.shell import ShellTask

with Flow(name, schedule=schedule) as flow:
  task(command='./do-the-thing')
  flow.register(project_name=project_name)

# do-the-thing
#!/usr/bin/env python3
from prefect.client import Secret

secret_key = Secret("SECRET_KEY")
os.environ['SECRET_KEY'] = secret_key.get()

Unexpected error: TimeoutError()

PrefectResult

@task(result=PrefectResult())
def add(x, y=1):
    if random.random() > 0.7:
        raise Exception('I failed on purpose')

    return x + y


with Flow("my handled flow!") as flow:
    first_result = add.map(list(range(10)), y=unmapped(2))

LocalResult

apply_map

@task
def double_number(number):
  return number * 2

with Flow("name") as flow:
  number = 4
  double_number(number)

12:10:11 prefect.CloudTaskRunner	Task 'run_simulator[54]': Starting task run...
12:10:11 prefect.run_simulator[54]	Starting simulator run
12:10:11 prefect.run_simulator[54]	cusip_list [{'secmnem': 'FNMMA3057', 'cusip': '31418CMF8'}]
12:10:11 prefect.run_simulator[54]	Loading model 'cf621134-8c36-446a-96b5-7ecde88a33e2'
12:10:22 prefect.run_simulator[54]	Simulating pool {'secmnem': 'FNMMA3057', 'cusip': '31418CMF8'}
12:10:31 prefect.run_simulator[54]	Number of replicates 6
12:11:59 prefect.CloudTaskRunner	Task 'run_simulator[54]': finished task run for task with final state: 'Success'

11:55:34 prefect.CloudTaskRunner	Task 'run_simulator[6]': Starting task run...
11:55:35 prefect.run_simulator[6]	Starting simulator run
11:55:35 prefect.run_simulator[6]	cusip_list [{'secmnem': 'FNMMA3774', 'cusip': '31418DFQ0'}]
11:55:35 prefect.run_simulator[6]	Loading model 'c410358f-4612-4aef-8f12-e9a3642711de'
11:56:23 prefect.run_simulator[6]	Simulating pool {'secmnem': 'FNMMA3774', 'cusip': '31418DFQ0'}
11:56:36 prefect.run_simulator[6]	Number of replicates 3
11:57:12 prefect.CloudTaskRunner	Task 'run_simulator[6]': finished task run for task with final state: 'Success'
12:06:17 prefect.CloudTaskRunner	Task 'run_simulator[6]': Starting task run...
12:06:17 prefect.run_simulator[6]	Starting simulator run
12:06:17 prefect.run_simulator[6]	cusip_list [{'secmnem': 'FNMMA3774', 'cusip': '31418DFQ0'}]
12:06:17 prefect.run_simulator[6]	Loading model '45322fce-d452-4340-9e06-e7bcc2775b84'
12:06:27 prefect.run_simulator[6]	Simulating pool {'secmnem': 'FNMMA3774', 'cusip': '31418DFQ0'}
12:06:40 prefect.run_simulator[6]	Number of replicates 3
12:07:15 prefect.CloudTaskRunner	Task 'run_simulator[6]': finished task run for task with final state: 'Success'

flow.run(executor=DaskExecutor(
        cluster_class=LocalCluster, cluster_kwargs={"n_workers": 2, "threads_per_worker": 1}))

with Flow("Files downloader") as flow:
        files = get_files()
        downloaded_files = download_file.map(files)
        import_file.map(downloaded_files)

import_file

download_file

LOOP

results = transform_data.map(call_api())

call_api

LOOP

transform_data

backend

flow.run

prefect diagnostics

{
  "config_overrides": {
    "cloud": {
      "agent": {
        "auth_token": true
      }
    },
    "context": {
      "secrets": false
    }
  },
  "env_vars": [
    "PREFECT__CLOUD__AGENT__AUTH_TOKEN"
  ],
  "system_information": {
    "platform": "Linux-4.19.104-microsoft-standard-x86_64-with-glibc2.29",
    "prefect_version": "0.13.4",
    "python_version": "3.8.1"
  }
}

for dataset in dataset_config['datasets']:

print('starting flow for dataset' + dataset['dataset_name'])

flow.run(dataset=dataset['dataset_name'])  # runs this flow on its schedule

@task

@task

task

Retry-After

retry_delay

some_task.map(df.groupby("id"))

KeyError: 'Column not found: 0'

ad-agent_1              |   File "/usr/local/lib/python3.7/site-packages/requests/models.py", line 941, in raise_for_status
ad-agent_1              |     raise HTTPError(http_error_msg, response=self)
ad-agent_1              | requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http://ad-prefect-apollo:4200/>

agent

PREFECT__SERVER__ENDPOINT: <http://ad-prefect-apollo:4200>

LocalDaskExecutor()

flow.run(..

LocalEnvironment(executor=LocalDaskExecutor())

with Flow("test", schedule=schedule, environment=LocalEnvironment(executor=LocalDaskExecutor())) as flow:

@task
def generate():
    return [x for x in range(0,40)]

@task
def log_sleep(x):
    logger = prefect.context.get('logger')
    time.sleep(5 + x)
    <http://logger.info|logger.info>(x)
    return x * x

@task
def collect(lst):
    logger = prefect.context.get('logger')
    <http://logger.info|logger.info>(lst)

schedule = Schedule(clocks=[DatesClock([pendulum.now() + timedelta(seconds=5)])])

with Flow("test", schedule=schedule, environment=LocalEnvironment(executor=LocalDaskExecutor())) as flow:

    nums = generate()

    results = log_sleep.map(nums)

    x = collect(results)

flow.run()

from prefect import task, Flow, Parameter


@task(log_stdout=True)
def say_hello(name):
    print("Hello, {}!".format(name))


with Flow("My First Flow") as flow:
    name = Parameter('name')
    say_hello(name)

flow.register('Test')

func_task

# Here's an example of what I'm thinking
corpus_file = 'corpus.txt'
@task(target=corpus_file, result=LocalResult())
def write_sentence_corpus(sentences):
    with open(corpus_file, 'w') as fp:
        for sent in sentences:
            fp.write(sent + '\n')

with Flow("text_model") as f:
    sentences = extract_sentences()
    corpus = write_sentence_corpus(sentences)
    ...
    # Do more modelling