Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_canonicalized_resource(self, request):
uri_path = request.path.split('?')[0]
# for emulator, use the DEV_ACCOUNT_NAME instead of DEV_ACCOUNT_SECONDARY_NAME
# as this is how the emulator works
if self.is_emulated and uri_path.find(DEV_ACCOUNT_SECONDARY_NAME) == 1:
# only replace the first instance
uri_path = uri_path.replace(DEV_ACCOUNT_SECONDARY_NAME, DEV_ACCOUNT_NAME, 1)
return '/' + self.account_name + uri_path
def _get_canonicalized_resource(self, request):
uri_path = request.path.split('?')[0]
# for emulator, use the DEV_ACCOUNT_NAME instead of DEV_ACCOUNT_SECONDARY_NAME
# as this is how the emulator works
if self.is_emulated and uri_path.find(DEV_ACCOUNT_SECONDARY_NAME) == 1:
# only replace the first instance
uri_path = uri_path.replace(DEV_ACCOUNT_SECONDARY_NAME, DEV_ACCOUNT_NAME, 1)
return '/' + self.account_name + uri_path
if context.location_mode == LocationMode.PRIMARY:
context.location_mode = LocationMode.SECONDARY
# if targeting the emulator (with path style), change path instead of host
if context.is_emulated:
# replace the first instance of primary account name with the secondary account name
context.request.path = context.request.path.replace(DEV_ACCOUNT_NAME, DEV_ACCOUNT_SECONDARY_NAME, 1)
else:
context.request.host = context.request.host_locations.get(context.location_mode)
else:
context.location_mode = LocationMode.PRIMARY
# if targeting the emulator (with path style), change path instead of host
if context.is_emulated:
# replace the first instance of secondary account name with the primary account name
context.request.path = context.request.path.replace(DEV_ACCOUNT_SECONDARY_NAME, DEV_ACCOUNT_NAME, 1)
else:
context.request.host = context.request.host_locations.get(context.location_mode)
'''
A function which sets the next host location on the request, if applicable.
:param ~azure.storage.models.RetryContext context:
The retry context containing the previous host location and the request
to evaluate and possibly modify.
'''
if len(context.request.host_locations) > 1:
# If there's more than one possible location, retry to the alternative
if context.location_mode == LocationMode.PRIMARY:
context.location_mode = LocationMode.SECONDARY
# if targeting the emulator (with path style), change path instead of host
if context.is_emulated:
# replace the first instance of primary account name with the secondary account name
context.request.path = context.request.path.replace(DEV_ACCOUNT_NAME, DEV_ACCOUNT_SECONDARY_NAME, 1)
else:
context.request.host = context.request.host_locations.get(context.location_mode)
else:
context.location_mode = LocationMode.PRIMARY
# if targeting the emulator (with path style), change path instead of host
if context.is_emulated:
# replace the first instance of secondary account name with the primary account name
context.request.path = context.request.path.replace(DEV_ACCOUNT_SECONDARY_NAME, DEV_ACCOUNT_NAME, 1)
else:
context.request.host = context.request.host_locations.get(context.location_mode)
self.account_key = account_key
self.sas_token = sas_token
self.token_credential = token_credential
self.protocol = protocol or DEFAULT_PROTOCOL
self.is_emulated = is_emulated
if is_emulated:
self.account_name = DEV_ACCOUNT_NAME
self.protocol = 'http'
# Only set the account key if a sas_token is not present to allow sas to be used with the emulator
self.account_key = DEV_ACCOUNT_KEY if not self.sas_token else None
emulator_endpoint = _EMULATOR_ENDPOINTS[service] if custom_domain is None else custom_domain
self.primary_endpoint = '{}/{}'.format(emulator_endpoint, DEV_ACCOUNT_NAME)
self.secondary_endpoint = '{}/{}'.format(emulator_endpoint, DEV_ACCOUNT_SECONDARY_NAME)
else:
# Strip whitespace from the key
if self.account_key:
self.account_key = self.account_key.strip()
endpoint_suffix = endpoint_suffix or SERVICE_HOST_BASE
# Setup the primary endpoint
if custom_domain:
parsed_url = urlparse(custom_domain)
# Trim any trailing slashes from the path
path = parsed_url.path.rstrip('/')
self.primary_endpoint = parsed_url.netloc + path
self.protocol = self.protocol if parsed_url.scheme is '' else parsed_url.scheme