Python Examples of botocore.vendored.requests.get (2023)

def process_record(payload): # construct ddb entry from data ddb_item = { 'mid': str(payload['entities']['media'][0]['id']), 'tid': payload['id'], 'media': payload['entities']['media'][0]['media_url'], 'text': payload['text'], 'sn': payload['user']['screen_name'], 'mentions': ['@'+mention['screen_name'] for mention in payload['entities']['user_mentions'] if TWITTER_SN[1:] not in mention['screen_name']], 'ts': int(time.time()) } # grab image from twitter resp = requests.get(ddb_item['media']+":large") # store unprocessed image in s3 bucket for rekognition to process unprocessed_bucket.put_object(Body=resp.content, Key=str(ddb_item['mid']), ACL='public-read') ddb.put_item(Item=ddb_item) 

def _get_response(self, full_url, headers, timeout): try: response = self._session.get(full_url, headers=headers, timeout=timeout) if response.status_code != 200: raise MetadataRetrievalError( error_msg="Received non 200 response (%s) from ECS metadata: %s" % (response.status_code, response.text)) try: return json.loads(response.text) except ValueError: raise MetadataRetrievalError( error_msg=("Unable to parse JSON returned from " "ECS metadata: %s" % response.text)) except RETRYABLE_HTTP_ERRORS as e: error_msg = ("Received error when attempting to retrieve " "ECS metadata: %s" % e) raise MetadataRetrievalError(error_msg=error_msg) 

def _get_response(self, full_url, headers, timeout): try: response = self._session.get(full_url, headers=headers, timeout=timeout) if response.status_code != 200: raise MetadataRetrievalError( error_msg="Received non 200 response (%s) from ECS metadata: %s" % (response.status_code, response.text)) try: return json.loads(response.text) except ValueError: raise MetadataRetrievalError( error_msg=("Unable to parse JSON returned from " "ECS metadata: %s" % response.text)) except RETRYABLE_HTTP_ERRORS as e: error_msg = ("Received error when attempting to retrieve " "ECS metadata: %s" % e) raise MetadataRetrievalError(error_msg=error_msg) 

def handler(event, context): try: if event['RequestType'] == 'Create': # Test Integration print 'Getting all pets' response = requests.get(event['ResourceProperties']['IntegrationEndpoint']) print "Status code: " + str(response.status_code) if response.status_code != 200: raise Exception('Error: Status code received is not 200') elif event['RequestType'] == 'Update': pass elif event['RequestType'] == 'Delete': pass cfnresponse.send(event, context, cfnresponse.SUCCESS, {}, '') except: print traceback.print_exc() cfnresponse.send(event, context, cfnresponse.FAILED, {}, '') 

def _get_response(self, full_url, headers, timeout): try: response = self._session.get(full_url, headers=headers, timeout=timeout) if response.status_code != 200: raise MetadataRetrievalError( error_msg="Received non 200 response (%s) from ECS metadata: %s" % (response.status_code, response.text)) try: return json.loads(response.text) except ValueError: raise MetadataRetrievalError( error_msg=("Unable to parse JSON returned from " "ECS metadata: %s" % response.text)) except RETRYABLE_HTTP_ERRORS as e: error_msg = ("Received error when attempting to retrieve " "ECS metadata: %s" % e) raise MetadataRetrievalError(error_msg=error_msg) 

def get_members(zip): parts = [] # get all the path prefixes for name in zip.namelist(): # only check files (not directories) if not name.endswith('/'): # keep list of path elements (minus filename) parts.append(name.split('/')[:-1]) # now find the common path prefix (if any) prefix = os.path.commonprefix(parts) if prefix: # re-join the path elements prefix = '/'.join(prefix) + '/' # get the length of the common prefix offset = len(prefix) # now re-set the filenames for zipinfo in zip.infolist(): name = zipinfo.filename # only check files (not directories) if len(name) > offset: # remove the common prefix zipinfo.filename = name[offset:] yield zipinfo 

def figure_out_schedule(s): # Purpose here is to find the schedule id if given a human readable name # fingers crossed that this regex holds for awhile. "PXXXXXX" if re.match('^P[a-zA-Z0-9]{6}', s): return s global PD_API_KEY headers = { 'Accept': 'application/vnd.pagerduty+json;version=2', 'Authorization': 'Token token={token}'.format(token=PD_API_KEY) } url = 'https://api.pagerduty.com/schedules/' payload = {} payload['query'] = s # If there is no override, then check the schedule directly r = requests.get(url, headers=headers, params=payload) try: # This is fragile. fuzzy search may not do what you want sid = r.json()['schedules'][0]['id'] except IndexError: logger.debug("Schedule Not Found for: {}".format(s)) sid = None return sid 

def metadata_toc(self): if self._metadata_toc is None: res = requests.get(self.mds_url) res.raise_for_status() jwt_header = jwt.get_unverified_header(res.content) assert jwt_header["alg"] == "ES256" cert = x509.load_der_x509_certificate(jwt_header["x5c"][0].encode(), cryptography.hazmat.backends.default_backend()) self._metadata_toc = jwt.decode(res.content, key=cert.public_key(), algorithms=["ES256"]) return self._metadata_toc 

def metadata_for_key_id(self, key_id): for e in self.metadata_toc["entries"]: if key_id in e.get("attestationCertificateKeyIdentifiers", []): break else: raise KeyError("No metadata found for key ID {}".format(key_id)) res = requests.get(e["url"]) res.raise_for_status() return json.loads(base64.b64decode(res.content).decode()) 

def validate_record(payload): if ( TWITTER_SN in payload.get('text', '') and payload.get('entities', {}).get('media') and 'RT' not in payload.get('text') ): return True return False 

def get_service_module_name(service_model): """Returns the module name for a service This is the value used in both the documentation and client class name """ name = service_model.metadata.get( 'serviceAbbreviation', service_model.metadata.get( 'serviceFullName', service_model.service_name)) name = name.replace('Amazon', '') name = name.replace('AWS', '') name = re.sub('\W+', '', name) return name 

def _get_request(self, url, timeout, num_attempts=1): for i in range(num_attempts): try: response = requests.get(url, timeout=timeout) except RETRYABLE_HTTP_ERRORS as e: logger.debug("Caught exception while trying to retrieve " "credentials: %s", e, exc_info=True) else: if response.status_code == 200: return response raise _RetriesExceededError() 

def instance_cache(func): """Method decorator for caching method calls to a single instance. **This is not a general purpose caching decorator.** In order to use this, you *must* provide an ``_instance_cache`` attribute on the instance. This decorator is used to cache method calls. The cache is only scoped to a single instance though such that multiple instances will maintain their own cache. In order to keep things simple, this decorator requires that you provide an ``_instance_cache`` attribute on your instance. """ func_name = func.__name__ @functools.wraps(func) def _cache_guard(self, *args, **kwargs): cache_key = (func_name, args) if kwargs: kwarg_items = tuple(sorted(kwargs.items())) cache_key = (func_name, args, kwarg_items) result = self._instance_cache.get(cache_key) if result is not None: return result result = func(self, *args, **kwargs) self._instance_cache[cache_key] = result return result return _cache_guard 

def switch_host_with_param(request, param_name): """Switches the host using a parameter value from a JSON request body""" request_json = json.loads(request.data.decode('utf-8')) if request_json.get(param_name): new_endpoint = request_json[param_name] _switch_hosts(request, new_endpoint) 

def get_bucket_region(self, bucket, response): """ There are multiple potential sources for the new region to redirect to, but they aren't all universally available for use. This will try to find region from response elements, but will fall back to calling HEAD on the bucket if all else fails. :param bucket: The bucket to find the region for. This is necessary if the region is not available in the error response. :param response: A response representing a service request that failed due to incorrect region configuration. """ # First try to source the region from the headers. service_response = response[1] response_headers = service_response['ResponseMetadata']['HTTPHeaders'] if 'x-amz-bucket-region' in response_headers: return response_headers['x-amz-bucket-region'] # Next, check the error body region = service_response.get('Error', {}).get('Region', None) if region is not None: return region # Finally, HEAD the bucket. No other choice sadly. try: response = self._client.head_bucket(Bucket=bucket) headers = response['ResponseMetadata']['HTTPHeaders'] except ClientError as e: headers = e.response['ResponseMetadata']['HTTPHeaders'] region = headers.get('x-amz-bucket-region', None) return region 

def set_request_url(self, params, context, **kwargs): endpoint = context.get('signing', {}).get('endpoint', None) if endpoint is not None: params['url'] = _get_new_endpoint(params['url'], endpoint, False) 

def redirect_from_cache(self, params, context, **kwargs): """ This handler retrieves a given bucket's signing context from the cache and adds it into the request context. """ bucket = params.get('Bucket') signing_context = self._cache.get(bucket) if signing_context is not None: context['signing'] = signing_context else: context['signing'] = {'bucket': bucket} 

def get_service_module_name(service_model): """Returns the module name for a service This is the value used in both the documentation and client class name """ name = service_model.metadata.get( 'serviceAbbreviation', service_model.metadata.get( 'serviceFullName', service_model.service_name)) name = name.replace('Amazon', '') name = name.replace('AWS', '') name = re.sub('\W+', '', name) return name 

def instance_cache(func): """Method decorator for caching method calls to a single instance. **This is not a general purpose caching decorator.** In order to use this, you *must* provide an ``_instance_cache`` attribute on the instance. This decorator is used to cache method calls. The cache is only scoped to a single instance though such that multiple instances will maintain their own cache. In order to keep things simple, this decorator requires that you provide an ``_instance_cache`` attribute on your instance. """ func_name = func.__name__ @functools.wraps(func) def _cache_guard(self, *args, **kwargs): cache_key = (func_name, args) if kwargs: kwarg_items = tuple(sorted(kwargs.items())) cache_key = (func_name, args, kwarg_items) result = self._instance_cache.get(cache_key) if result is not None: return result result = func(self, *args, **kwargs) self._instance_cache[cache_key] = result return result return _cache_guard 

def switch_host_with_param(request, param_name): """Switches the host using a parameter value from a JSON request body""" request_json = json.loads(request.data.decode('utf-8')) if request_json.get(param_name): new_endpoint = request_json[param_name] _switch_hosts(request, new_endpoint) 

def get_bucket_region(self, bucket, response): """ There are multiple potential sources for the new region to redirect to, but they aren't all universally available for use. This will try to find region from response elements, but will fall back to calling HEAD on the bucket if all else fails. :param bucket: The bucket to find the region for. This is necessary if the region is not available in the error response. :param response: A response representing a service request that failed due to incorrect region configuration. """ # First try to source the region from the headers. service_response = response[1] response_headers = service_response['ResponseMetadata']['HTTPHeaders'] if 'x-amz-bucket-region' in response_headers: return response_headers['x-amz-bucket-region'] # Next, check the error body region = service_response.get('Error', {}).get('Region', None) if region is not None: return region # Finally, HEAD the bucket. No other choice sadly. try: response = self._client.head_bucket(Bucket=bucket) headers = response['ResponseMetadata']['HTTPHeaders'] except ClientError as e: headers = e.response['ResponseMetadata']['HTTPHeaders'] region = headers.get('x-amz-bucket-region', None) return region 

def redirect_from_cache(self, params, context, **kwargs): """ This handler retrieves a given bucket's signing context from the cache and adds it into the request context. """ bucket = params.get('Bucket') signing_context = self._cache.get(bucket) if signing_context is not None: context['signing'] = signing_context else: context['signing'] = {'bucket': bucket} 

def api_trigger(event, context): gateway_url = os.getenv("PY_API_GATEWAY_URL") context.iopipe.metric("gateway_url", gateway_url or "") if gateway_url is not None: response = requests.get(gateway_url) context.iopipe.metric("response_status", response.status_code) 

def auto_http(event, context): requests.get("https://www.iopipe.com") client = boto3.client("s3") client.list_buckets() 

def auto_http_30k(event, context): for _ in range(30000): requests.get( "https://www.google.com/images/branding/googlelogo/2x/googlelogo_color_272x92dp.png" ) 

def api_trigger(event, context): gateway_url = os.getenv("PY_API_GATEWAY_URL") context.iopipe.metric("gateway_url", gateway_url or "") if gateway_url is not None: response = requests.get(gateway_url) context.iopipe.metric("response_status", response.status_code) 

def auto_http(event, context): requests.get("https://www.iopipe.com") client = boto3.client("s3") client.list_buckets() 

def post_method(data=None, content_type=None): if not data: data = request.body.read().decode('utf-8') if not content_type: content_type = request.content_type if 'x-amz-sns-message-type' not in request.headers.keys(): raise Exception('missing headers') if request.headers['x-amz-sns-message-type'] != 'SubscriptionConfirmation': return url = json.loads(data)['SubscribeURL'] requests.get(url) return 

def is_json_value_header(shape): """Determines if the provided shape is the special header type jsonvalue. :type shape: botocore.shape :param shape: Shape to be inspected for the jsonvalue trait. :return: True if this type is a jsonvalue, False otherwise :rtype: Bool """ return (hasattr(shape, 'serialization') and shape.serialization.get('jsonvalue', False) and shape.serialization.get('location') == 'header' and shape.type_name == 'string') 

def get_service_module_name(service_model): """Returns the module name for a service This is the value used in both the documentation and client class name """ name = service_model.metadata.get( 'serviceAbbreviation', service_model.metadata.get( 'serviceFullName', service_model.service_name)) name = name.replace('Amazon', '') name = name.replace('AWS', '') name = re.sub(r'\W+', '', name) return name