my prefect server setup
prefect-metrics.waow.tech
python
orchestration
redesign pds-records config for proper UI form experience
- records: list[dict] replaces record: dict (batch create support)
- updates: dict replaces overloaded record field for update action
- position ordering and [action]-scoped descriptions on all fields
- action renders as dropdown, fields are logically grouped
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+61
-18
+61
-18
flows/pds_records.py
+61
-18
flows/pds_records.py
···
25
25
update_record,
26
26
)
27
27
28
+
Action = Literal["list", "delete", "create", "update"]
29
+
28
30
29
31
class PdsRecordsConfig(BaseModel):
30
-
action: Literal["list", "delete", "create", "update"]
31
-
collection: str = Field(description="e.g. network.cosmik.connection")
32
-
repo: str | None = Field(default=None, description="target repo (default: authenticated user)")
33
-
record: dict[str, Any] | None = Field(default=None, description="for create: the record body")
34
-
uri: str | None = Field(default=None, description="for update/delete: specific record AT-URI")
35
-
rkey_filter: str | None = Field(default=None, description="for delete: regex filter on rkey (None = all)")
36
-
dry_run: bool = Field(default=True, description="delete safety default")
32
+
"""PDS record operations. Fields are shared across actions — unused fields are ignored."""
33
+
34
+
action: Action = Field(
35
+
description="list: enumerate records. delete: remove matching records. create: write new records. update: patch an existing record.",
36
+
json_schema_extra=dict(position=0),
37
+
)
38
+
collection: str = Field(
39
+
description="NSID of the collection, e.g. network.cosmik.connection",
40
+
json_schema_extra=dict(position=1),
41
+
)
42
+
repo: str | None = Field(
43
+
default=None,
44
+
description="target repo DID or handle (default: authenticated user)",
45
+
json_schema_extra=dict(position=2),
46
+
)
47
+
48
+
# create / update
49
+
records: list[dict[str, Any]] = Field(
50
+
default_factory=list,
51
+
description="[create] record bodies to create. ignored for other actions.",
52
+
json_schema_extra=dict(position=3),
53
+
)
54
+
uri: str | None = Field(
55
+
default=None,
56
+
description="[update] AT-URI of the record to update",
57
+
json_schema_extra=dict(position=4),
58
+
)
59
+
updates: dict[str, Any] | None = Field(
60
+
default=None,
61
+
description="[update] fields to merge into the existing record",
62
+
json_schema_extra=dict(position=5),
63
+
)
64
+
65
+
# delete
66
+
rkey_filter: str | None = Field(
67
+
default=None,
68
+
description="[delete] regex filter on rkey — only matching records are deleted. omit to match all.",
69
+
json_schema_extra=dict(position=6),
70
+
)
71
+
dry_run: bool = Field(
72
+
default=True,
73
+
description="[delete] preview what would be deleted without actually deleting",
74
+
json_schema_extra=dict(position=7),
75
+
)
37
76
38
77
39
78
async def _paginate_all(
···
91
130
92
131
93
132
@task(cache_policy=NONE)
94
-
async def create_pds_record(client: AsyncClient, config: PdsRecordsConfig) -> dict[str, str]:
95
-
"""Create a record, return URI + CID."""
96
-
if not config.record:
97
-
raise ValueError("config.record is required for create action")
133
+
async def create_pds_records(client: AsyncClient, config: PdsRecordsConfig) -> list[dict[str, str]]:
134
+
"""Create one or more records, return URIs + CIDs."""
135
+
if not config.records:
136
+
raise ValueError("config.records is required for create action")
98
137
99
-
resp = await create_record(client, config.collection, config.record)
100
-
print(f"created {resp.uri} (cid={resp.cid})")
101
-
return {"uri": resp.uri, "cid": resp.cid}
138
+
results = []
139
+
for record in config.records:
140
+
resp = await create_record(client, config.collection, record)
141
+
print(f" created {resp.uri} (cid={resp.cid})")
142
+
results.append({"uri": resp.uri, "cid": resp.cid})
143
+
print(f"created {len(results)} records")
144
+
return results
102
145
103
146
104
147
@task(cache_policy=NONE)
···
106
149
"""Update a record at the given URI."""
107
150
if not config.uri:
108
151
raise ValueError("config.uri is required for update action")
109
-
if not config.record:
110
-
raise ValueError("config.record is required for update action")
152
+
if not config.updates:
153
+
raise ValueError("config.updates is required for update action")
111
154
112
-
resp = await update_record(client, config.uri, config.record)
155
+
resp = await update_record(client, config.uri, config.updates)
113
156
print(f"updated {resp.uri} (cid={resp.cid})")
114
157
return {"uri": resp.uri, "cid": resp.cid}
115
158
···
134
177
case "delete":
135
178
await delete_pds_records(client, config)
136
179
case "create":
137
-
await create_pds_record(client, config)
180
+
await create_pds_records(client, config)
138
181
case "update":
139
182
await update_pds_record(client, config)
140
183