-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathdocument_report.py
78 lines (67 loc) · 2.11 KB
/
document_report.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
""""""
from typing import TYPE_CHECKING, List, Union
from indico import PagedRequest
from indico.filters import DocumentReportFilter
from indico.types import BaseType
from indico.types.document_report import DocumentReport
if TYPE_CHECKING: # pragma: no cover
from typing import Optional, Union
from indico.typing import AnyDict, Payload
class _DocumentReportList(BaseType):
submissions: List[DocumentReport]
class GetDocumentReport(PagedRequest["List[DocumentReport]"]):
"""
Query to generate a Document Report, otherwise known as a log of past submissions.
"""
query = """
query SubmissionsLog($filters: SubmissionLogFilter, $limit: Int, $after: Int, $allSubmissions: Boolean){
submissionsLog(filters: $filters, limit: $limit, after: $after, allSubmissions: $allSubmissions){
submissions{
datasetId
workflowId
status
createdAt
createdBy
updatedAt
updatedBy
completedAt
errors
retrieved
submissionId
filesDeleted
inputFiles{
id
filename
filepath
submissionId
fileSize
numPages
}
}
pageInfo{
startCursor
endCursor
hasNextPage
aggregateCount
}
}
}
"""
def __init__(
self,
filters: "Optional[Union[AnyDict, DocumentReportFilter]]" = None,
limit: "Optional[int]" = None,
all_submissions: bool = False,
):
variables = {
"filters": filters,
"limit": limit,
"allSubmissions": all_submissions,
}
super().__init__(self.query, variables=variables)
def process_response(
self, response: "Payload", _: "Optional[str]" = None
) -> "List[DocumentReport]":
return _DocumentReportList(
**super().parse_payload(response)["submissionsLog"]
).submissions