-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdspacerequest.py
More file actions
174 lines (128 loc) · 9.2 KB
/
Copy pathdspacerequest.py
File metadata and controls
174 lines (128 loc) · 9.2 KB
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
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
from dataobjects import AuthData
import requests
from requests_toolbelt import MultipartEncoder
from pathlib import Path
import mimetypes
class DspaceRequest:
def __init__(self, authData: AuthData, dspaceRestURL) -> None:
self.authData = authData
self.dspaceRestURL = dspaceRestURL
def auth_status(self):
auth_req = requests.get(self.dspaceRestURL + "/api/authn/status", headers={"Accept":"application/json", "Authorization": self.authData.bearer_jwt})
if auth_req.status_code == requests.codes.ok:
return auth_req
auth_req.raise_for_status()
def save_auth(self):
resp = self.auth_status()
if "DSPACE-XSRF-TOKEN" in resp.headers:
self.authData.csrf_token = resp.headers["DSPACE-XSRF-TOKEN"]
self.authData.auth_cookie = resp.cookies
if "Authorization" in resp.headers:
self.authData.bearer_jwt = resp.headers["Authorization"].split(" ")[1]
def success(self, req: requests.Response, status_code):
if req.status_code == status_code:
self.save_auth()
return req.json()
req.raise_for_status()
def success_true(self, req: requests.Response, status_code):
if req.status_code == status_code:
self.save_auth()
return True
req.raise_for_status()
def ok_response(self, req: requests.Response):
return self.success(req, requests.codes.ok)
def dspace_get_no_cookie(self, url, req_headers):
return self.success(requests.get(url, headers=req_headers), requests.codes.ok)
def dspace_get(self, url, req_headers, req_cookies):
return self.success(requests.get(url, headers=req_headers, cookies=req_cookies), requests.codes.ok)
def dspace_create_post(self, url, req_headers, req_cookies, req_data):
return self.success(requests.post(url, headers=req_headers, cookies=req_cookies, data=req_data), requests.codes.created)
class PublicDspaceRequest(DspaceRequest):
_self = None
def __new__(cls, *args, **kwargs):
if cls._self is None:
cls._self = super().__new__(cls)
return cls._self
def __init__(self, dspaceRestURL, auth_data: AuthData):
super().__init__(auth_data, dspaceRestURL)
def get_metadata_schemas(self):
return self.dspace_get_no_cookie(f"{self.dspaceRestURL}/api/core/metadataschemas", {"Accept":"application/json"})
def get_metadata_schema_fields(self, prefix, page):
return self.dspace_get_no_cookie(f"{self.dspaceRestURL}/api/core/metadatafields/search/bySchema?schema={prefix}&page={page}", {"Accept": "application/json"})
class DspaceCommunityRequest(DspaceRequest):
_self = None
def __new__(cls, *args, **kwargs):
if cls._self is None:
cls._self = super().__new__(cls)
return cls._self
def __init__(self, dspaceRestURL: str, auth_data: AuthData) -> None:
super().__init__(auth_data, dspaceRestURL)
def top_communities(self): #return json
return self.dspace_get(f"{self.dspaceRestURL}/api/core/communities/search/top", {"Accept":"application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
def sub_communities(self, community_uuid, page_number = 0):
return self.dspace_get(f"{self.dspaceRestURL}/api/core/communities/{community_uuid}/subcommunities?page={page_number}", {"Accept":"application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
def sub_collections(self, community_uuid, page_number = 0):
return self.dspace_get(f"{self.dspaceRestURL}/api/core/communities/{community_uuid}/collections?page={page_number}", {"Accept":"application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
class DspaceItemRequest(DspaceRequest):
_self = None
def __new__(cls, *args, **kwargs):
if cls._self is None:
cls._self = super().__new__(cls)
return cls._self
def __init__(self, dspaceRestURL: str, auth_data: AuthData):
super().__init__(auth_data, dspaceRestURL)
def get_item(self, item_uuid):
return self.dspace_get(f"{self.dspaceRestURL}/api/core/items/{item_uuid}", {"Accept": "application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
def item_owning_collection(self, item_uuid):
return self.dspace_get(f"{self.dspaceRestURL}/api/core/items/{item_uuid}/owningCollection", {"Accept":"application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
def item_bundles(self, item_uuid):
return self.dspace_get(f"{self.dspaceRestURL}/api/core/items/{item_uuid}/bundles", {"Accept": "application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
def new_item(self, item_json, owning_collection_uuid):
return self.dspace_create_post(f"{self.dspaceRestURL}/api/core/items?owningCollection={owning_collection_uuid}", {"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token, 'Content-Type': 'application/json', 'Accept': 'application/json'}, self.authData.auth_cookie, item_json)
def put_item(self, item_uuid, item_json):
req = requests.put(f"{self.dspaceRestURL}/api/core/items/{item_uuid}", headers={"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token, "Content-Type": "application/json", "Accept": "application/json"}, cookies=self.authData.auth_cookie, data=item_json)
return self.ok_response(req)
def patch_item(self, item_uuid, patch_json):
req = requests.patch(f"{self.dspaceRestURL}/api/core/items/{item_uuid}", headers={"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token, "Content-Type": "application/json", "Accept": "application/json"}, cookies=self.authData.auth_cookie, data=patch_json)
return self.ok_response(req)
class DspaceBundleRequest(DspaceRequest):
_self = None
def __new__(cls, *args, **kwargs):
if cls._self is None:
cls._self = super().__new__(cls)
return cls._self
def __init__(self, dspaceRestURL: str, auth_data: AuthData) -> None:
super().__init__(auth_data, dspaceRestURL)
def has_primary_bitstream(self, bundle_uuid) -> bool:
req = requests.get(f"{self.dspaceRestURL}/api/core/bundles/{bundle_uuid}/primaryBitstream", headers={"Authorization": self.authData.bearer_jwt}, cookies=self.authData.auth_cookie)
return req.status_code == requests.codes.ok
def delete_primary_bitstream_flag(self, bundle_uuid):
req = requests.delete(f"{self.dspaceRestURL}/api/core/bundles/{bundle_uuid}/primaryBitstream", headers={"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token}, cookies=self.authData.auth_cookie)
return self.success_true(req, requests.codes.no_content)
def get_bitstreams(self, bundle_uuid, page):
return self.dspace_get(f"{self.dspaceRestURL}/api/core/bundles/{bundle_uuid}/bitstreams?page={page}", {"Accept": "application/json", "Authorization": self.authData.bearer_jwt}, self.authData.auth_cookie)
def new_bundle(self, bundle_json, item_uuid):
return self.dspace_create_post(f"{self.dspaceRestURL}/api/core/items/{item_uuid}/bundles", {"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token, "Content-Type": "application/json", "Accept": "application/json"}, self.authData.auth_cookie, bundle_json)
def add_primary_bitstream(self, bundle_uuid, bitstream_uuid):
req = requests.post(f"{self.dspaceRestURL}/api/core/bundles/{bundle_uuid}/primaryBitstream", headers={"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token, "Content-Type": "text/uri-list"}, cookies=self.authData.auth_cookie, data=f"{self.dspaceRestURL}/api/core/bitstreams/{bitstream_uuid}")
return self.success_true(req, requests.codes.created)
class DspaceBitstreamRequest(DspaceRequest):
_self = None
def __new__(cls, *args, **kwargs):
if cls._self is None:
cls._self = super().__new__(cls)
return cls._self
def __init__(self, dspaceRestURL: str, auth_data: AuthData) -> None:
super().__init__(auth_data, dspaceRestURL)
def delete_bitstream(self, bitstream_uuid):
req = requests.delete(f"{self.dspaceRestURL}/api/core/bitstreams/{bitstream_uuid}", headers={"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token}, cookies=self.authData.auth_cookie)
return self.success_true(req, requests.codes.no_content)
def new_bitstream(self, bundle_uuid, file: Path, file_metadata_json):
m_encoder = MultipartEncoder(
fields={
'properties': file_metadata_json,
'file': (file.name, open(file, "rb"), mimetypes.types_map[file.suffix.lower()])
}
)
req = requests.post(f"{self.dspaceRestURL}/api/core/bundles/{bundle_uuid}/bitstreams", headers={"Authorization": self.authData.bearer_jwt, "X-XSRF-TOKEN": self.authData.csrf_token, "Accept": "application/json", "Content-Type": m_encoder.content_type}, cookies=self.authData.auth_cookie, data=m_encoder)
return self.success(req, requests.codes.created)