Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create_collections(self, payload, last_modified=None):
"""
Create new Zotero collections
Accepts one argument, a list of dicts containing the following keys:
'name': the name of the collection
'parentCollection': OPTIONAL, the parent collection to which you wish to add this
"""
# no point in proceeding if there's no 'name' key
for item in payload:
if "name" not in item:
raise ze.ParamNotPassed("The dict you pass must include a 'name' key")
# add a blank 'parentCollection' key if it hasn't been passed
if "parentCollection" not in item:
item["parentCollection"] = ""
headers = {"Zotero-Write-Token": token()}
if last_modified is not None:
headers["If-Unmodified-Since-Version"] = str(last_modified)
headers.update(self.default_headers())
self._check_backoff()
req = requests.post(
url=self.endpoint
+ "/{t}/{u}/collections".format(t=self.library_type, u=self.library_id),
headers=headers,
data=json.dumps(payload),
)
self.request = req
try:
def _verify(self, payload):
"""
ensure that all files to be attached exist
open()'s better than exists(), cos it avoids a race condition
"""
if not payload: # Check payload has nonzero length
raise ze.ParamNotPassed
for templt in payload:
if os.path.isfile(str(self.basedir.joinpath(templt["filename"]))):
try:
# if it is a file, try to open it, and catch the error
with open(str(self.basedir.joinpath(templt["filename"]))):
pass
except IOError:
raise ze.FileDoesNotExist(
"The file at %s couldn't be opened or found."
% str(self.basedir.joinpath(templt["filename"]))
)
# no point in continuing if the file isn't a file
else:
raise ze.FileDoesNotExist(
"The file at %s couldn't be opened or found."
% str(self.basedir.joinpath(templt["filename"]))
def _build_query(self, query_string):
"""
Set request parameters. Will always add the user ID if it hasn't
been specifically set by an API method
"""
try:
query = quote(query_string.format(
u=self.library_id,
t=self.library_type))
except KeyError as err:
raise ze.ParamNotPassed(
'There\'s a request parameter missing: %s' % err)
# Add the URL parameters and the user key, if necessary
if not self.url_params:
self.add_parameters()
query = '%s?%s' % (query, self.url_params)
return query
def _validate(self, conditions):
""" Validate saved search conditions, raising an error if any contain invalid operators """
allowed_keys = set(self.searchkeys)
operators_set = set(self.operators.keys())
for condition in conditions:
if set(condition.keys()) != allowed_keys:
raise ze.ParamNotPassed(
"Keys must be all of: %s" % ", ".join(self.searchkeys)
)
if condition.get("operator") not in operators_set:
raise ze.ParamNotPassed(
"You have specified an unknown operator: %s"
% condition.get("operator")
)
# dict keys of allowed operators for the current condition
permitted_operators = self.conditions_operators.get(
condition.get("condition")
)
# transform these into values
permitted_operators_list = set(
[self.operators.get(op) for op in permitted_operators]
)
if condition.get("operator") not in permitted_operators_list:
def _build_query(self, query_string, no_params=False):
"""
Set request parameters. Will always add the user ID if it hasn't
been specifically set by an API method
"""
try:
query = quote(query_string.format(u=self.library_id, t=self.library_type))
except KeyError as err:
raise ze.ParamNotPassed("There's a request parameter missing: %s" % err)
# Add the URL parameters and the user key, if necessary
if no_params is False:
if not self.url_params:
self.add_parameters()
query = "%s?%s" % (query, self.url_params)
return query
def create_collection(self, payload):
"""
Create a new Zotero collection
Accepts one argument, a dict containing the following keys:
'name': the name of the collection
'parent': OPTIONAL, the parent collection to which you wish to add this
"""
# no point in proceeding if there's no 'name' key
for item in payload:
if 'name' not in item:
raise ze.ParamNotPassed(
"The dict you pass must include a 'name' key")
# add a blank 'parent' key if it hasn't been passed
if not 'parentCollection' in item:
payload['parentCollection'] = ''
headers = {
'Zotero-Write-Token': token(),
}
headers.update(self.default_headers())
req = requests.post(
url=self.endpoint
+ '/{t}/{u}/collections'.format(
t=self.library_type,
u=self.library_id),
headers=headers,
data=json.dumps(payload))
try: