KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_async_generator_result method in localstack

Best Python code snippet using localstack_python

http2_server.py

Source:http2_server.py

...77class HTTPErrorResponse(Exception):78    def __init__(self, *args, code=None, **kwargs):79        super(HTTPErrorResponse, self).__init__(*args, **kwargs)80        self.code = code81def get_async_generator_result(result):82    gen, headers = result, {}83    if isinstance(result, tuple) and len(result) >= 2:84        gen, headers = result[:2]85    if not isinstance(gen, (collections.abc.Generator, collections.abc.AsyncGenerator)):86        return87    return gen, headers88def run_server(port, bind_address, handler=None, asynchronous=True, ssl_creds=None):89    ensure_event_loop()90    app = Quart(__name__)91    app.config["MAX_CONTENT_LENGTH"] = 256 * 1024 * 1024  # 256 MB request payload limit92    @app.route("/", methods=HTTP_METHODS, defaults={"path": ""})93    @app.route("/<path:path>", methods=HTTP_METHODS)94    async def index(path=None):95        response = await make_response("{}")96        if handler:97            data = await request.get_data()98            try:99                result = await run_sync(handler, request, data)100                if isinstance(result, Exception):101                    raise result102            except Exception as e:103                LOG.warning(104                    "Error in proxy handler for request %s %s: %s %s"105                    % (request.method, request.url, e, traceback.format_exc())106                )107                response.status_code = 500108                if isinstance(e, HTTPErrorResponse):109                    response.status_code = e.code or response.status_code110                return response111            if result is not None:112                # check if this is an async generator (for HTTP2 push event responses)113                async_gen = get_async_generator_result(result)114                if async_gen:115                    return async_gen116                # prepare and return regular response117                is_chunked = uses_chunked_encoding(result)118                result_content = result.content or ""119                response = await make_response(result_content)120                response.status_code = result.status_code121                if is_chunked:122                    response.headers.pop("Content-Length", None)123                result.headers.pop("Server", None)124                result.headers.pop("Date", None)125                headers = {k: str(v).replace("\n", r"\n") for k, v in result.headers.items()}126                response.headers.update(headers)127                # set multi-value headers...

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.