records = get_records()
for record in records:
    tags = []
    record_id = get_record_id_if_exist()
    if record_id == None:
        for tag in record["tags"]:
            tag_id = get_record_tag_if_exist()
            if tag_id == None:
                tag_id = create_tag()
            tags.append(tag_id)
        record_id = create_open_data(record['open_data'], tag_list)
    version_id = create_version(record_id, record['version'])

apply_map

y

from prefect import Flow, task, case, apply_map
from prefect.tasks.control_flow import merge

@task
def inc(x,y):
    return x + 1

@task
def negate(x,y):
    return -x

@task
def is_even(x,y):
    return x % 2 == 0

def inc_or_negate(x, y):
    cond = is_even(x,y)
    with case(cond, True):
        res1 = inc(x,y)
    with case(cond, False):
        res2 = negate(x,y)
    return merge(res1, res2)

for x in [1,2,3,4]:
    for y in ["A","B","C","D"]:
        inc_or_negate(x, y)

@task(nout=2)
def cross_product(x_list, y_list):
    res = []
    for x in x_list:
        for y in y_list:
            res.append((x,y))
    res_x = [_[0] for _ in res]
    res_y = [_[1] for _ in res]
    return res_x, res_y

with Flow("apply-map example") as flow:
    x, y = cross_product(range(4), ["A","B","C", "D"])
    result = apply_map(inc_or_negate, x, y)

flow.run()

x

y