https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> ```@flow(log_prints=True) async def l...
# ask-marvin
t
@Marvin
Copy code
@flow(log_prints=True)
async def load_wso_industry_codes(as_of_date: date | None = None) -> None:
    as_of_date = date.today() - timedelta(days=1) if as_of_date is None else as_of_date
    issuer_ids = get_issuer_ids(as_of_date)
    # print(f'done getting issuer id from db. Count: {len(issuer_ids)}')
    sem = asyncio.Semaphore(20) # avoid blasting WSO server, 20 is sufficiently fast
    async with httpx.AsyncClient(timeout=30) as client:
        token = await get_wso_token(client)
        client.headers['Authorization'] = f"Bearer {token}"

        issuers = await work(sem, client, issuer_ids, wait_for=[issuer_ids])

        print('saving industry codes')
        await save_industry_codes(as_of_date, issuers, wait_for=[issuers])
why prefect dependency graph is showing a dependency arrow betwen save_industry_codes and get_issuer_ids
m
✅ thought for 208.2 seconds
Because Prefect is likely showing the upstream lineage, not just the single 
wait_for
edge you wrote. In your flow:
Copy code
python
issuer_ids = get_issuer_ids(as_of_date)
issuers = await work(sem, client, issuer_ids, wait_for=[issuer_ids])
await save_industry_codes(as_of_date, issuers, wait_for=[issuers])
the dependency chain is:
Copy code
text
get_issuer_ids -> work -> save_industry_codes
So even though 
save_industry_codes
only directly waits on `issuers`/`work`, 
work
itself depends on 
get_issuer_ids
because 
issuer_ids
is passed into it. A few important details: * 
wait_for=[issuer_ids]
creates an explicit dependency from 
work
to 
get_issuer_ids
* passing 
issuer_ids
as an argument to 
work(...)
also creates a dependency automatically * then 
save_industry_codes(..., issuers, wait_for=[issuers])
depends on 
work
So if the UI draws an arrow from 
get_issuer_ids
to 
save_industry_codes
, it's usually representing that 
save_industry_codes
is indirectly upstream-constrained by 
get_issuer_ids
, even though that is not the direct argument you passed to 
save_industry_codes
. One subtle thing in your snippet:
Copy code
python
issuer_ids = get_issuer_ids(as_of_date)
If 
get_issuer_ids
is a Prefect task, this returns a future-like object, not the raw data, and Prefect tracks that as a task dependency when you pass it downstream. Also, this part is redundant:
Copy code
python
issuers = await work(sem, client, issuer_ids, wait_for=[issuer_ids])
If 
issuer_ids
is already passed as an argument to 
work
, Prefect already knows 
work
depends on 
get_issuer_ids
. You typically do not need 
wait_for=[issuer_ids]
too. Likewise:
Copy code
python
await save_industry_codes(as_of_date, issuers, wait_for=[issuers])
If 
issuers
is an input argument to 
save_industry_codes
, the dependency is already inferred. 
wait_for=[issuers]
is usually unnecessary unless you want ordering without data passing. So the cleaner version is probably:
Copy code
python
@flow(log_prints=True)
async def load_wso_industry_codes(as_of_date: date | None = None) -> None:
    as_of_date = date.today() - timedelta(days=1) if as_of_date is None else as_of_date
    issuer_ids = get_issuer_ids(as_of_date)

    sem = asyncio.Semaphore(20)
    async with httpx.AsyncClient(timeout=30) as client:
        token = await get_wso_token(client)
        client.headers["Authorization"] = f"Bearer {token}"

        issuers = await work(sem, client, issuer_ids)

        print("saving industry codes")
        await save_industry_codes(as_of_date, issuers)
In short: Prefect sees 
save_industry_codes
as depending on 
get_issuer_ids
through 
work
, so the graph may show that lineage. Docs: * Prefect tasks * Prefect flows If you want, I can also explain the difference between data dependencies vs 
wait_for
-only dependencies in Prefect graphs.
t
@Marvin i have removed the wait_for and the dependency graph still show the same, I don't want to show save_industry_codes dependent on get_issuer_ids
m
✅ thought for 12.1 seconds
That makes sense — and the reason is that 
save_industry_codes
is still data-dependent on 
get_issuer_ids
through 
work
. Even without 
wait_for
, Prefect infers dependencies from task inputs:
Copy code
text
get_issuer_ids -> work -> save_industry_codes
Since 
save_industry_codes
receives 
issuers
, and 
issuers
comes from 
work
, and 
work
received 
issuer_ids
, Prefect can still show 
get_issuer_ids
as part of the upstream chain. If you don't want that dependency shown, you need to break the Prefect task/future lineage. Options: 1. Resolve the upstream result into plain Python data before passing it onward If 
issuer_ids
is a task future, Prefect tracks it. If you materialize it outside task dependency passing, you may reduce the graph lineage. For example, depending on how your tasks are defined:
Copy code
python
issuer_ids = await get_issuer_ids(as_of_date)
issuer_ids_data = issuer_ids
issuers = await work(sem, client, issuer_ids_data)
await save_industry_codes(as_of_date, issuers)
But if 
work
is a Prefect task and 
issuers
is still its task result, then 
save_industry_codes
will still depend on 
work
. 2. Move the intermediate logic out of Prefect tasks If 
work
is a Prefect task, then its output carries tracked lineage. If you make 
work
a regular async Python function instead of a task, then only 
save_industry_codes
remains as a task boundary. Example pattern:
Copy code
python
issuer_ids = await get_issuer_ids(as_of_date)
issuers = await work_regular_python_function(sem, client, issuer_ids)
await save_industry_codes(as_of_date, issuers)
Then 
save_industry_codes
should only depend on 
get_issuer_ids
if 
issuer_ids
is also passed into it or embedded in a tracked future path. 3. Make 
save_industry_codes
a regular function instead of a task If your main concern is the graph visualization, only Prefect task/flow runs appear as tracked nodes. A plain function won't show this dependency structure. 4. Detach by returning plain data from the flow context before the next task call Sometimes people do this by awaiting/resolving futures and then passing raw values, but if both producer and consumer are still Prefect tasks in the same flow, Prefect often still preserves the lineage between task runs. So the practical answer is: As long as 
save_industry_codes
consumes the output of a Prefect task that ultimately came from 
get_issuer_ids
, Prefect will consider it part of the same dependency chain. If your real goal is: * 
save_industry_codes
should run after 
work
* but the UI should not visually imply direct dependency on 
get_issuer_ids
then the cleanest approach is usually: * keep 
work
and 
save_industry_codes
as plain Python functions, or * combine some of these steps into a single Prefect task boundary For example:
Copy code
python
@task
async def fetch_issuers_and_transform(as_of_date, sem, client):
    issuer_ids = await get_issuer_ids(as_of_date)
    return await work(sem, client, issuer_ids)

@flow
async def load_wso_industry_codes(as_of_date: date | None = None) -> None:
    as_of_date = date.today() - timedelta(days=1) if as_of_date is None else as_of_date
    sem = asyncio.Semaphore(20)

    async with httpx.AsyncClient(timeout=30) as client:
        token = await get_wso_token(client)
        client.headers["Authorization"] = f"Bearer {token}"

        issuers = await fetch_issuers_and_transform(as_of_date, sem, client)
        await save_industry_codes(as_of_date, issuers)
Now the graph is simpler because 
get_issuer_ids
is hidden inside one task boundary. So the short version: You cannot keep Prefect-aware task lineage and also hide that upstream dependency in the graph. To hide it, you need to collapse task boundaries or use plain Python functions for some steps. If you want, I can help refactor your exact snippet into a version that produces the cleaner graph you want.
3 Views
Open in Slack
PreviousNext
Powered by