my prefect server setup
prefect-metrics.waow.tech
python
orchestration
fix: disable cache policy on pds-records tasks (AsyncClient not serializable)
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+5
-4
+5
-4
flows/pds_records.py
+5
-4
flows/pds_records.py
···
15
15
16
16
from prefect import flow, task
17
17
from prefect.blocks.system import Secret
18
+
from prefect.cache_policies import NONE
18
19
19
20
from pdsx._internal.auth import login
20
21
from pdsx._internal.operations import (
···
51
52
return all_records
52
53
53
54
54
-
@task
55
+
@task(cache_policy=NONE)
55
56
async def list_pds_records(client: AsyncClient, config: PdsRecordsConfig) -> list[dict[str, Any]]:
56
57
"""List all records in a collection, print count + sample."""
57
58
records = await _paginate_all(client, config.collection, config.repo)
···
63
64
return records
64
65
65
66
66
-
@task
67
+
@task(cache_policy=NONE)
67
68
async def delete_pds_records(client: AsyncClient, config: PdsRecordsConfig) -> int:
68
69
"""List → optional rkey filter → dry_run preview or delete each."""
69
70
records = await _paginate_all(client, config.collection, config.repo)
···
89
90
return deleted
90
91
91
92
92
-
@task
93
+
@task(cache_policy=NONE)
93
94
async def create_pds_record(client: AsyncClient, config: PdsRecordsConfig) -> dict[str, str]:
94
95
"""Create a record, return URI + CID."""
95
96
if not config.record:
···
100
101
return {"uri": resp.uri, "cid": resp.cid}
101
102
102
103
103
-
@task
104
+
@task(cache_policy=NONE)
104
105
async def update_pds_record(client: AsyncClient, config: PdsRecordsConfig) -> dict[str, str]:
105
106
"""Update a record at the given URI."""
106
107
if not config.uri: