Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use flask_error_response_json method in localstack

Best Python code snippet using localstack_python

aws_responses.py

Source:aws_responses.py

...39    code = 40440    def __init__(self, message=None):41        message = message or "The given resource cannot be found"42        super(ResourceNotFoundException, self).__init__("ResourceNotFoundException", message)43def flask_error_response_json(44    msg: str, code: Optional[int] = 500, error_type: Optional[str] = "InternalFailure"45):46    result = {47        "Type": "User" if code < 500 else "Server",48        "message": msg,49        "__type": error_type,50    }51    headers = {"x-amzn-errortype": error_type}52    # Note: don't use flask's make_response(..) or jsonify(..) here as they53    # can lead to "RuntimeError: working outside of application context".54    return FlaskResponse(json.dumps(result), status=code, headers=headers)55def requests_error_response_json(message, code=500, error_type="InternalFailure"):56    response = flask_error_response_json(message, code=code, error_type=error_type)57    return flask_to_requests_response(response)58def requests_error_response_xml(59    message: str,60    code: Optional[int] = 400,61    code_string: Optional[str] = "InvalidParameter",62    service: Optional[str] = None,63    xmlns: Optional[str] = None,64):65    response = RequestsResponse()66    xmlns = xmlns or "http://%s.amazonaws.com/doc/2010-03-31/" % service67    response._content = """<ErrorResponse xmlns="{xmlns}"><Error>68        <Type>Sender</Type>69        <Code>{code_string}</Code>70        <Message>{message}</Message>71        </Error><RequestId>{req_id}</RequestId>72        </ErrorResponse>""".format(73        xmlns=xmlns, message=message, code_string=code_string, req_id=short_uid()74    )75    response.status_code = code76    return response77def to_xml(data: dict, memberize: bool = True) -> ET.Element:78    """Generate XML element hierarchy out of dict. Wraps list items in <member> tags by default"""79    if not isinstance(data, dict) or len(data.keys()) != 1:80        raise Exception("Expected data to be a dict with a single root element")81    def _to_xml(parent_el: ET.Element, data_rest) -> None:82        if isinstance(data_rest, list):83            for i in data_rest:84                member_el = ET.SubElement(parent_el, "member") if memberize else parent_el85                _to_xml(member_el, i)86        elif isinstance(data_rest, dict):87            for key in data_rest:88                value = data_rest[key]89                curr_el = ET.SubElement(parent_el, key)90                _to_xml(curr_el, value)91        elif isinstance(data_rest, str):92            parent_el.text = data_rest93        elif any(94            isinstance(data_rest, i) for i in [bool, str, int, float]95        ):  # limit types for text serialization96            parent_el.text = str(data_rest)97        else:98            if data_rest is not None:  # None is just ignored and omitted99                raise Exception(f"Unexpected type for value encountered: {type(data_rest)}")100    root_key = list(data.keys())[0]101    root = ET.Element(root_key)102    _to_xml(root, data[root_key])103    return root104def requests_response_xml(action, response, xmlns=None, service=None, memberize=True):105    xmlns = xmlns or "http://%s.amazonaws.com/doc/2010-03-31/" % service106    response = json_safe(response)107    response = {"{action}Result".format(action=action): response}108    response = ET.tostring(to_xml(response, memberize=memberize), short_empty_elements=True)109    response = to_str(response)110    result = (111        """112        <{action}Response xmlns="{xmlns}">113            {response}114        </{action}Response>115        """116    ).strip()117    result = result.format(action=action, xmlns=xmlns, response=response)118    result = requests_response(result)119    return result120def requests_error_response_xml_signature_calculation(121    message,122    string_to_sign=None,123    signature=None,124    expires=None,125    code=400,126    code_string="AccessDenied",127    aws_access_token="temp",128):129    response = RequestsResponse()130    response_template = """<?xml version="1.0" encoding="UTF-8"?>131        <Error>132            <Code>{code_string}</Code>133            <Message>{message}</Message>134            <RequestId>{req_id}</RequestId>135            <HostId>{host_id}</HostId>136        </Error>""".format(137        message=message,138        code_string=code_string,139        req_id=short_uid(),140        host_id=short_uid(),141    )142    parsed_response = xmltodict.parse(response_template)143    response.status_code = code144    if signature and string_to_sign or code_string == "SignatureDoesNotMatch":145        bytes_signature = binascii.hexlify(bytes(signature, encoding="utf-8"))146        parsed_response["Error"]["Code"] = code_string147        parsed_response["Error"]["AWSAccessKeyId"] = aws_access_token148        parsed_response["Error"]["StringToSign"] = string_to_sign149        parsed_response["Error"]["SignatureProvided"] = signature150        parsed_response["Error"]["StringToSignBytes"] = "{}".format(bytes_signature.decode("utf-8"))151        set_response_content(response, xmltodict.unparse(parsed_response))152    if expires and code_string == "AccessDenied":153        server_time = datetime.datetime.utcnow().isoformat()[:-4]154        expires_isoformat = datetime.datetime.fromtimestamp(int(expires)).isoformat()[:-4]155        parsed_response["Error"]["Code"] = code_string156        parsed_response["Error"]["Expires"] = "{}Z".format(expires_isoformat)157        parsed_response["Error"]["ServerTime"] = "{}Z".format(server_time)158        set_response_content(response, xmltodict.unparse(parsed_response))159    if not signature and not expires and code_string == "AccessDenied":160        set_response_content(response, xmltodict.unparse(parsed_response))161    if response._content:162        return response163def flask_error_response_xml(164    message: str,165    code: Optional[int] = 500,166    code_string: Optional[str] = "InternalFailure",167    service: Optional[str] = None,168    xmlns: Optional[str] = None,169):170    response = requests_error_response_xml(171        message, code=code, code_string=code_string, service=service, xmlns=xmlns172    )173    return requests_to_flask_response(response)174def requests_error_response(175    req_headers: Dict,176    message: Union[str, bytes],177    code: int = 500,178    error_type: str = "InternalFailure",179    service: str = None,180    xmlns: str = None,181):182    is_json = is_json_request(req_headers)183    if is_json:184        return requests_error_response_json(message=message, code=code, error_type=error_type)185    return requests_error_response_xml(186        message, code=code, code_string=error_type, service=service, xmlns=xmlns187    )188def is_json_request(req_headers: Dict) -> bool:189    ctype = req_headers.get("Content-Type", "")190    accept = req_headers.get("Accept", "")191    return "json" in ctype or "json" in accept192def is_invalid_html_response(headers, content) -> bool:193    content_type = headers.get("Content-Type", "")194    return "text/html" in content_type and not str_startswith_ignore_case(content, "<!doctype html")195def raise_exception_if_error_response(response):196    if not is_response_obj(response):197        return198    if response.status_code < 400:199        return200    content = "..."201    try:202        content = truncate(to_str(response.content or ""))203    except Exception:204        pass  # ignore if content has non-printable bytes205    raise Exception("Received error response (code %s): %s" % (response.status_code, content))206def is_response_obj(result, include_lambda_response=False):207    types = (RequestsResponse, FlaskResponse)208    if include_lambda_response:209        types += (LambdaResponse,)210    return isinstance(result, types)211def get_response_payload(response, as_json=False):212    result = (213        response.content214        if isinstance(response, RequestsResponse)215        else response.data216        if isinstance(response, FlaskResponse)217        else None218    )219    result = "" if result is None else result220    if as_json:221        result = result or "{}"222        result = json.loads(to_str(result))223    return result224def requests_response(content, status_code=200, headers=None):225    if headers is None:226        headers = {}227    resp = RequestsResponse()228    headers = CaseInsensitiveDict(dict(headers or {}))229    if isinstance(content, dict):230        content = json.dumps(content)231        if not headers.get(HEADER_CONTENT_TYPE):232            headers[HEADER_CONTENT_TYPE] = APPLICATION_JSON233    resp._content = content234    resp.status_code = int(status_code)235    # Note: update headers (instead of assigning directly), to ensure we're using a case-insensitive dict236    resp.headers.update(headers)237    return resp238def request_response_stream(stream, status_code=200, headers=None):239    if headers is None:240        headers = {}241    resp = RequestsResponse()242    resp.raw = stream243    resp.status_code = int(status_code)244    # Note: update headers (instead of assigning directly), to ensure we're using a case-insensitive dict245    resp.headers.update(headers or {})246    return resp247def flask_to_requests_response(r):248    return requests_response(r.data, status_code=r.status_code, headers=r.headers)249def requests_to_flask_response(r):250    return FlaskResponse(r.content, status=r.status_code, headers=dict(r.headers))251def flask_not_found_error(msg=None):252    msg = msg or "The specified resource doesnt exist."253    return flask_error_response_json(msg, code=404, error_type="ResourceNotFoundException")254def response_regex_replace(response, search, replace):255    content = re.sub(search, replace, to_str(response._content), flags=re.DOTALL | re.MULTILINE)256    set_response_content(response, content)257def set_response_content(response, content, headers=None):258    if isinstance(content, dict):259        content = json.dumps(json_safe(content))260    elif isinstance(content, RequestsResponse):261        response.status_code = content.status_code262        content = content.content263    response._content = content or ""264    response.headers.update(headers or {})265    response.headers["Content-Length"] = str(len(response._content))266def make_requests_error(*args, **kwargs):267    return flask_to_requests_response(flask_error_response_xml(*args, **kwargs))...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.