21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319 | @dataclass
@async_to_sync
class FormGroup(FormGroupMixin, FormGroupProtocol):
"""Dataclass representing a FormGroup."""
# Default states for listing FormData
DEFAULT_OWNER_STATES = [
"waiting_for_submission",
"submitted_waiting_for_review",
"accepted",
"rejected",
]
DEFAULT_REVIEWER_STATES = [
"submitted_waiting_for_review",
"accepted",
"rejected",
]
async def create_or_get_async(
self,
*,
synapse_client: Optional["Synapse"] = None,
) -> "FormGroup":
"""
Create or get a FormGroup with the provided name. This method is idempotent. If a group with the provided name already exists and the caller has ACCESS_TYPE.READ permission the existing FormGroup will be returned.
Arguments:
synapse_client: Optional Synapse client instance for authentication.
Returns:
A FormGroup object containing the details of the created group.
Examples: create a FormGroup
```python
from synapseclient import Synapse
from synapseclient.models import FormGroup
import asyncio
async def create_my_form_group():
syn = Synapse()
syn.login()
form_group = FormGroup(name="my_unique_form_group_name")
form_group = await form_group.create_or_get_async()
print(form_group)
asyncio.run(create_my_form_group())
```
"""
if not self.name:
raise ValueError("FormGroup 'name' must be provided to create a FormGroup.")
from synapseclient.api.form_services import create_form_group
response = await create_form_group(
synapse_client=synapse_client,
name=self.name,
)
return self.fill_from_dict(response)
def _validate_filter_by_state(
self,
filter_by_state: List[str],
as_reviewer: bool = False,
) -> None:
"""
Validate filter_by_state values.
Arguments:
filter_by_state: List of str values to validate.
as_reviewer: If True, uses the POST POST /form/data/list/reviewer endpoint to review submission. If False (default), use POST /form/data/list endpoint to list only FormData owned by the caller.
"""
if not filter_by_state:
return
valid_string_values = [
"waiting_for_submission",
"submitted_waiting_for_review",
"accepted",
"rejected",
]
if as_reviewer:
valid_string_values.remove("waiting_for_submission")
for state in filter_by_state:
if state not in valid_string_values:
raise ValueError(
f"Invalid state: {state}. Valid values are: {', '.join(valid_string_values)}"
)
def _convert_state_enum_strings(
self,
state_enum_list: List[str],
) -> List[StateEnum]:
"""
Convert list of state enum strings to StateEnum values.
Arguments:
state_enum_list: List of StateEnum values as string.
Returns:
List of string values corresponding to the StateEnum.
"""
state_enum_mapping = {
"waiting_for_submission": StateEnum.WAITING_FOR_SUBMISSION,
"submitted_waiting_for_review": StateEnum.SUBMITTED_WAITING_FOR_REVIEW,
"accepted": StateEnum.ACCEPTED,
"rejected": StateEnum.REJECTED,
}
return [
state_enum_mapping.get(state)
for state in state_enum_list
if state in state_enum_mapping
]
@skip_async_to_sync
async def list_async(
self,
*,
filter_by_state: Optional[List[str]] = None,
synapse_client: Optional["Synapse"] = None,
as_reviewer: bool = False,
) -> AsyncGenerator["FormData", None]:
"""
List FormData objects in a FormGroup.
Arguments:
filter_by_state: list of StateEnum to filter the results.
When as_reviewer=False (default), valid values are:
- waiting_for_submission
- submitted_waiting_for_review
- accepted
- rejected
When as_reviewer=True, valid values are:
- submitted_waiting_for_review
- accepted
- rejected
Note: waiting_for_submission is NOT allowed when as_reviewer=True.
synapse_client: The Synapse client to use for the request.
as_reviewer: If True, uses the POST POST /form/data/list/reviewer endpoint to review submission. If False (default), use POST /form/data/list endpoint to list only FormData owned by the caller.
Yields:
FormData objects matching the request.
Raises:
ValueError: If group_id is not set or filter_by_state contains invalid values.
Examples: List your own form data
```python
from synapseclient import Synapse
from synapseclient.models import FormGroup
import asyncio
async def list_my_form_data():
syn = Synapse()
syn.login()
form_group = await FormGroup(name="test").create_or_get_async()
async for form_data in form_group.list_async(
filter_by_state=["submitted_waiting_for_review"]
):
status = form_data.submission_status
print(f"Form name: {form_data.name}")
print(f"State: {status.state.value}")
print(f"Submitted on: {status.submitted_on}")
asyncio.run(list_my_form_data())
```
Examples: List all form data as a reviewer
```python
from synapseclient import Synapse
from synapseclient.models import FormGroup
import asyncio
async def list_my_form_data():
syn = Synapse()
syn.login()
form_group = await FormGroup(name="test").create_or_get_async()
async for form_data in form_group.list_async(as_reviewer=True):
status = form_data.submission_status
print(f"Form name: {form_data.name}")
print(f"State: {status.state.value}")
print(f"Submitted on: {status.submitted_on}")
asyncio.run(list_my_form_data())
```
"""
from synapseclient.api import list_form_data
if not self.group_id:
raise ValueError(
"'group_id' must be provided to list FormData within a form group."
)
if not filter_by_state:
if as_reviewer:
filter_by_state = self.DEFAULT_REVIEWER_STATES
else:
filter_by_state = self.DEFAULT_OWNER_STATES
self._validate_filter_by_state(
filter_by_state=filter_by_state,
as_reviewer=as_reviewer,
)
filter_by_state_enum = self._convert_state_enum_strings(
state_enum_list=filter_by_state
)
gen = list_form_data(
synapse_client=synapse_client,
group_id=self.group_id,
filter_by_state=filter_by_state_enum,
as_reviewer=as_reviewer,
)
async for item in gen:
yield FormData().fill_from_dict(item)
def list(
self,
*,
filter_by_state: Optional[List[str]] = None,
synapse_client: Optional["Synapse"] = None,
as_reviewer: bool = False,
) -> Generator["FormData", None, None]:
"""
List FormData objects in a FormGroup.
Arguments:
filter_by_state: Optional list of StateEnum to filter the results.
When as_reviewer=False (default), valid values are:
- waiting_for_submission
- submitted_waiting_for_review
- accepted
- rejected
When as_reviewer=True, valid values are:
- submitted_waiting_for_review
- accepted
- rejected
Note: waiting_for_submission is NOT allowed when as_reviewer=True.
as_reviewer: If True, uses the reviewer endpoint (requires READ_PRIVATE_SUBMISSION
permission). If False (default), lists only FormData owned by the caller.
synapse_client: The Synapse client to use for the request.
Yields:
FormData objects matching the request.
Raises:
ValueError: If group_id is not set or filter_by_state contains invalid values.
Examples: List your own form data
```python
from synapseclient.models import FormGroup
from synapseclient import Synapse
syn = Synapse()
syn.login()
form_group = FormGroup(name="test").create_or_get()
list_data = form_group.list(filter_by_state=["waiting_for_submission"], as_reviewer=False)
for form_data in list_data:
print(f"FormData ID: {form_data.form_data_id}, State: {form_data.submission_status.state.value}")
```
Examples: List all form data as a reviewer
```python
from synapseclient.models import FormGroup
from synapseclient import Synapse
syn = Synapse()
syn.login()
form_group = FormGroup(name="test").create_or_get()
list_data = form_group.list(as_reviewer=True)
for form_data in list_data:
print(f"FormData ID: {form_data.form_data_id}, State: {form_data.submission_status.state.value}")
```
"""
yield from wrap_async_generator_to_sync_generator(
async_gen_func=self.list_async,
synapse_client=synapse_client,
filter_by_state=filter_by_state,
as_reviewer=as_reviewer,
)
|