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_request(self, server_node):
self.url = "{0}/databases/{1}/indexes?{2}".format(server_node.url, server_node.database,
"name={0}".format(
Utils.quote_key(self._index_name, True)))
def create_request(self, server_node):
if self.key_or_keys is None:
raise ValueError("None Key is not valid")
path = "docs?"
if self.includes:
path += "".join("&include=" + Utils.quote_key(item) for item in self.includes)
# make get method handle a multi document requests in a single request
if isinstance(self.key_or_keys, list):
key_or_keys = collections.OrderedDict.fromkeys(self.key_or_keys)
if self.metadata_only:
path += "&metadataOnly=true"
# If it is too big, we drop to POST (note that means that we can't use the HTTP cache any longer)
if (sum(len(x) for x in key_or_keys)) > 1024:
self.method = "POST"
self.data = list(key_or_keys)
else:
path += "".join("&id=" + Utils.quote_key(item) for item in key_or_keys)
else:
path += "&id={0}".format(Utils.quote_key(self.key_or_keys))
def delete(self, key, etag=None):
if key is None:
raise ValueError("None Key is not valid")
if not isinstance(key, str):
raise ValueError("key must be {0}".format(type("")))
headers = {}
if etag is not None:
headers["If-None-Match"] = etag
key = Utils.quote_key(key)
path = "docs/{0}".format(key)
response = self._requests_handler.http_request_handler(path, "DELETE", headers=headers)
if response.status_code != 204:
raise exceptions.ErrorResponseException(response.json()["Error"])
def create_request(self, server_node):
path = "docs?id={0}".format(Utils.quote_key(self._document_id))
if self._skip_patch_if_change_vector_mismatch:
path += "&skipPatchIfChangeVectorMismatch=true"
if self._return_debug_information:
path += "&debug=true"
if self._test:
path += "&test=true"
if self._change_vector is not None:
self.headers = {"If-Match": "\"{0}\"".format(self._change_vector)}
self.url = "{0}/databases/{1}/{2}".format(server_node.url, server_node.database, path)
self.data = {"Patch": self._patch.to_json(),
"PatchIfMissing": self._patch_if_missing.to_json() if self._patch_if_missing else None}
def create_request(self, server_node):
if self.key is None:
raise ValueError("None Key is not valid")
if not isinstance(self.key, str):
raise ValueError("key must be {0}".format(type("")))
if self.change_vector is not None:
self.headers = {"If-Match": "\"{0}\"".format(self.change_vector)}
self.url = "{0}/databases/{1}/docs?id={2}".format(server_node.url, server_node.database,
Utils.quote_key(self.key))
@param index_entries_only: True if query results should contain only index entries.
:type bool
@return:json
:rtype:dict
"""
if not index_name:
raise ValueError("index_name cannot be None or empty")
if index_query is None:
raise ValueError("None query is invalid")
if not isinstance(index_query, IndexQuery):
raise ValueError("query must be IndexQuery type")
path = "indexes/{0}?&pageSize={1}".format(Utils.quote_key(index_name), index_query.page_size)
if index_query.default_operator is QueryOperator.AND:
path += "&operator={0}".format(index_query.default_operator.value)
if index_query.query:
path += "&query={0}".format(Utils.quote_key(index_query.query))
if index_query.sort_hints:
for hint in index_query.sort_hints:
path += "&{0}".format(hint)
if index_query.sort_fields:
for field in index_query.sort_fields:
path += "&sort={0}".format(field)
if index_query.fetch:
for item in index_query.fetch:
path += "&fetch={0}".format(item)
if metadata_only:
path += "&metadata-only=true"
if index_entries_only:
path += "&debug=entries"
if includes and len(includes) > 0:
path += "".join("&include=" + item for item in includes)
response = self._requests_handler.http_request_handler(path, "GET",
def create_request(self, server_node):
self.url = "{0}/databases/{1}/attachments?id={2}&name={3}".format(server_node.url, server_node.database,
Utils.quote_key(self._document_id),
Utils.quote_key(self._name))
if self._content_type:
self.url += "&contentType={0}".format(Utils.quote_key(self._content_type))
self.data = self._stream
if self._change_vector is not None:
self.headers = {"If-Match": "\"{0}\"".format(self._change_vector)}
def create_request(self, server_node):
self.url = "{0}/databases/{1}/indexes?name={2}".format(server_node.url, server_node.database,
Utils.quote_key(self._index_name))
def create_request(self, server_node):
self.url = "{0}/databases/{1}/attachments?id={2}&name={3}".format(
server_node.url, server_node.database, Utils.quote_key(self._document_id),
Utils.quote_key(self._name))
if self._attachment_type != AttachmentType.document:
self.method = "POST"
self.data = {"Type": str(self._attachment_type), "ChangeVector": self._change_vector}
path += "".join("&include=" + Utils.quote_key(item) for item in includes)
# make get method handle a multi document requests in a single request
if isinstance(key_or_keys, list):
key_or_keys = collections.OrderedDict.fromkeys(key_or_keys)
if metadata_only:
path += "&metadata-only=True"
# If it is too big, we drop to POST (note that means that we can't use the HTTP cache any longer)
if (sum(len(x) for x in key_or_keys)) > 1024:
method = "POST"
data = list(key_or_keys)
else:
path += "".join("&id=" + Utils.quote_key(item) for item in key_or_keys)
else:
path += "&id={0}".format(Utils.quote_key(key_or_keys))
response = self._requests_handler.http_request_handler(path, method, data=data,
force_read_from_master=force_read_from_master)
if response.status_code == 200:
response = response.json()
return response