Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_dict_keys_camel_case(original_obj, ignore_for_parent_keys=None):
if isinstance(original_obj, six.string_types):
return original_obj
if not isinstance(original_obj, abc.Mapping) and not isinstance(original_obj, abc.Iterable):
# Either a primitive or something we don't know how to deal with...given the entry point (from the output of
# json.loads, which should be a dict) more likely a primitive
return original_obj
if isinstance(original_obj, abc.Mapping):
new_dict = {}
for key, value in six.iteritems(original_obj):
camelized_key = camelize(key)
if ignore_for_parent_keys is not None and camelized_key in ignore_for_parent_keys:
new_dict[camelized_key] = value
else:
new_dict[camelized_key] = make_dict_keys_camel_case(value, ignore_for_parent_keys)
return new_dict
if isinstance(original_obj, abc.Iterable):
new_list = []
for obj in original_obj:
new_list.append(make_dict_keys_camel_case(obj, ignore_for_parent_keys))
return new_list
if extra_kwargs:
raise ValueError(
"list_report_types got unknown kwargs: {!r}".format(extra_kwargs))
query_params = {
"compartmentId": compartment_id,
"page": kwargs.get("page", missing)
}
query_params = {k: v for (k, v) in six.iteritems(query_params) if v is not missing and v is not None}
header_params = {
"accept": "application/json",
"content-type": "application/json",
"opc-request-id": kwargs.get("opc_request_id", missing)
}
header_params = {k: v for (k, v) in six.iteritems(header_params) if v is not missing and v is not None}
retry_strategy = self.retry_strategy
if kwargs.get('retry_strategy'):
retry_strategy = kwargs.get('retry_strategy')
if retry_strategy:
return retry_strategy.make_retrying_call(
self.base_client.call_api,
resource_path=resource_path,
method=method,
query_params=query_params,
header_params=header_params,
response_type="ReportTypeCollection")
else:
return self.base_client.call_api(
resource_path=resource_path,
path_params = {
"topicId": topic_id
}
path_params = {k: v for (k, v) in six.iteritems(path_params) if v is not missing}
for (k, v) in six.iteritems(path_params):
if v is None or (isinstance(v, six.string_types) and len(v.strip()) == 0):
raise ValueError('Parameter {} cannot be None, whitespace or empty string'.format(k))
header_params = {
"accept": "application/json",
"content-type": "application/json",
"opc-request-id": kwargs.get("opc_request_id", missing)
}
header_params = {k: v for (k, v) in six.iteritems(header_params) if v is not missing and v is not None}
retry_strategy = self.retry_strategy
if kwargs.get('retry_strategy'):
retry_strategy = kwargs.get('retry_strategy')
if retry_strategy:
return retry_strategy.make_retrying_call(
self.base_client.call_api,
resource_path=resource_path,
method=method,
path_params=path_params,
header_params=header_params,
response_type="NotificationTopic")
else:
return self.base_client.call_api(
resource_path=resource_path,
To have this operation explicitly not perform any retries, pass an instance of :py:class:`~oci.retry.NoneRetryStrategy`.
:return: A :class:`~oci.response.Response` object with data of type :class:`~oci.key_management.models.Key`
:rtype: :class:`~oci.response.Response`
"""
resource_path = "/20180608/keys/{keyId}/actions/cancelDeletion"
method = "POST"
# Don't accept unknown kwargs
expected_kwargs = [
"retry_strategy",
"if_match",
"opc_request_id",
"opc_retry_token"
]
extra_kwargs = [_key for _key in six.iterkeys(kwargs) if _key not in expected_kwargs]
if extra_kwargs:
raise ValueError(
"cancel_key_deletion got unknown kwargs: {!r}".format(extra_kwargs))
path_params = {
"keyId": key_id
}
path_params = {k: v for (k, v) in six.iteritems(path_params) if v is not missing}
for (k, v) in six.iteritems(path_params):
if v is None or (isinstance(v, six.string_types) and len(v.strip()) == 0):
raise ValueError('Parameter {} cannot be None, whitespace or empty string'.format(k))
header_params = {
"accept": "application/json",
Builds a JSON POST object.
If obj is str, int, float, bool, None, return directly.
If obj is datetime.datetime, datetime.date
convert to string in iso8601 format.
If obj is list, sanitize each element in the list.
If obj is dict, return the dict.
If obj is swagger model, return the properties dict.
:param obj: The data to serialize.
:return: The serialized form of data.
"""
types = (six.string_types, six.integer_types, float, bool, type(None))
declared_swagger_type_to_acceptable_python_types = {
'str': six.string_types,
'bool': bool,
'int': (float, six.integer_types),
'float': (float, six.integer_types)
}
# if there is a declared type for this obj, then validate that obj is of that type. None types (either None or the NONE_SENTINEL) are not validated but
# instead passed through
if declared_type and not self.is_none_or_none_sentinel(obj):
if declared_type.startswith('dict(') and not isinstance(obj, dict):
self.raise_type_error_serializing_model(field_name, obj, declared_type)
elif declared_type.startswith('list[') and not (isinstance(obj, list) or isinstance(obj, tuple)):
self.raise_type_error_serializing_model(field_name, obj, declared_type)
elif declared_type in self.complex_type_mappings:
# if its supposed to be one of our models, it can either be an instance of that model OR a dict
if not isinstance(obj, dict) and not isinstance(obj, self.complex_type_mappings[declared_type]):
self.raise_type_error_serializing_model(field_name, obj, declared_type)
# Inject date, host, and content-type if missing
request.headers.setdefault(
"date", email.utils.formatdate(usegmt=True))
request.headers.setdefault(
"host", six.moves.urllib.parse.urlparse(request.url).netloc)
if enforce_content_headers:
request.headers.setdefault("content-type", "application/json")
# Requests with a body need to send content-type,
# content-length, and x-content-sha256
if sign_body:
# TODO: does not handle streaming bodies (files, stdin)
body = request.body or ""
if isinstance(body, six.string_types):
body = body.encode("utf-8")
if "x-content-sha256" not in request.headers:
m = hashlib.sha256(body)
base64digest = base64.b64encode(m.digest())
base64string = base64digest.decode("utf-8")
request.headers["x-content-sha256"] = base64string
request.headers.setdefault("content-length", str(len(body)))
:return: A :class:`~oci.response.Response` object with data of type list of :class:`~oci.work_requests.models.WorkRequestError`
:rtype: :class:`~oci.response.Response`
"""
resource_path = "/workRequests/{workRequestId}/errors"
method = "GET"
# Don't accept unknown kwargs
expected_kwargs = [
"retry_strategy",
"limit",
"page",
"sort_order",
"opc_request_id"
]
extra_kwargs = [_key for _key in six.iterkeys(kwargs) if _key not in expected_kwargs]
if extra_kwargs:
raise ValueError(
"list_work_request_errors got unknown kwargs: {!r}".format(extra_kwargs))
path_params = {
"workRequestId": work_request_id
}
path_params = {k: v for (k, v) in six.iteritems(path_params) if v is not missing}
for (k, v) in six.iteritems(path_params):
if v is None or (isinstance(v, six.string_types) and len(v.strip()) == 0):
raise ValueError('Parameter {} cannot be None, whitespace or empty string'.format(k))
if 'sort_order' in kwargs:
sort_order_allowed_values = ["ASC", "DESC"]
def _validate_encryption_context(encryption_context):
if encryption_context is None:
return
if not isinstance(encryption_context, dict):
raise TypeError("encryption_context must be a dict")
invalid_keys = []
invalid_values = []
invalid_prefix_keys = []
for key, value in six.iteritems(encryption_context):
# keys and values must be str and not bytes
# in python 2 'bytes' is just an alias for 'str' so it is okay
if (str != bytes and isinstance(key, bytes)) or not isinstance(key, six.string_types):
invalid_keys.append(key)
elif key.startswith("oci-"):
invalid_prefix_keys.append(key)
if (str != bytes and isinstance(value, bytes)) or not isinstance(value, six.string_types):
invalid_values.append(value)
if invalid_keys or invalid_values:
raise TypeError(
"encryption_context must be a dict with string keys and string values. Invalid keys: {}. Invalid values: {}".format(
str(invalid_keys), str(invalid_values)
)
)
if invalid_prefix_keys:
raise ValueError(
}
path_params = {k: v for (k, v) in six.iteritems(path_params) if v is not missing}
for (k, v) in six.iteritems(path_params):
if v is None or (isinstance(v, six.string_types) and len(v.strip()) == 0):
raise ValueError('Parameter {} cannot be None, whitespace or empty string'.format(k))
header_params = {
"accept": "application/json",
"content-type": "application/json",
"if-match": kwargs.get("if_match", missing),
"opc-request-id": kwargs.get("opc_request_id", missing),
"opc-retry-token": kwargs.get("opc_retry_token", missing)
}
header_params = {k: v for (k, v) in six.iteritems(header_params) if v is not missing and v is not None}
retry_strategy = self.retry_strategy
if kwargs.get('retry_strategy'):
retry_strategy = kwargs.get('retry_strategy')
if retry_strategy:
if not isinstance(retry_strategy, retry.NoneRetryStrategy):
self.base_client.add_opc_retry_token_if_needed(header_params)
return retry_strategy.make_retrying_call(
self.base_client.call_api,
resource_path=resource_path,
method=method,
path_params=path_params,
header_params=header_params,
body=schedule_vault_deletion_details,
response_type="Vault")
To have this operation explicitly not perform any retries, pass an instance of :py:class:`~oci.retry.NoneRetryStrategy`.
:return: A :class:`~oci.response.Response` object with data of type None
:rtype: :class:`~oci.response.Response`
"""
resource_path = "/applianceExportJobs/{applianceExportJobId}/actions/changeCompartment"
method = "POST"
# Don't accept unknown kwargs
expected_kwargs = [
"retry_strategy",
"if_match",
"opc_request_id",
"opc_retry_token"
]
extra_kwargs = [_key for _key in six.iterkeys(kwargs) if _key not in expected_kwargs]
if extra_kwargs:
raise ValueError(
"change_appliance_export_job_compartment got unknown kwargs: {!r}".format(extra_kwargs))
path_params = {
"applianceExportJobId": appliance_export_job_id
}
path_params = {k: v for (k, v) in six.iteritems(path_params) if v is not missing}
for (k, v) in six.iteritems(path_params):
if v is None or (isinstance(v, six.string_types) and len(v.strip()) == 0):
raise ValueError('Parameter {} cannot be None, whitespace or empty string'.format(k))
header_params = {
"accept": "application/json",