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

How to use setup_quart_logging method in localstack

Best Python code snippet using localstack_python

http2_server.py

Source:http2_server.py

...19HTTP_METHODS = ['GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'PATCH']20# cache of SSL contexts (indexed by cert file names)21SSL_CONTEXTS = {}22SSL_LOCK = threading.RLock()23def setup_quart_logging():24    # set up loggers to avoid duplicate log lines in quart25    for name in ['quart.app', 'quart.serving']:26        log = logging.getLogger(name)27        log.setLevel(logging.INFO if config.DEBUG else logging.WARNING)28        for hdl in list(log.handlers):29            log.removeHandler(hdl)30def apply_patches():31    def InformationalResponse_init(self, *args, **kwargs):32        if kwargs.get('status_code') == 100 and not kwargs.get('reason'):33            # add missing "100 Continue" keyword which makes boto3 HTTP clients fail/hang34            kwargs['reason'] = 'Continue'35        InformationalResponse_init_orig(self, *args, **kwargs)36    InformationalResponse_init_orig = h11.InformationalResponse.__init__37    h11.InformationalResponse.__init__ = InformationalResponse_init38    # skip error logging for ssl.SSLError in hypercorn tcp_server.py39    async def _read_data(self) -> None:40        try:41            return await _read_data_orig(self)42        except Exception:43            await self.protocol.handle(Closed())44    _read_data_orig = tcp_server.TCPServer._read_data45    tcp_server.TCPServer._read_data = _read_data46    # avoid SSL context initialization errors when running multiple server threads in parallel47    def create_ssl_context(self, *args, **kwargs):48        with SSL_LOCK:49            key = '%s%s' % (self.certfile, self.keyfile)50            if key not in SSL_CONTEXTS:51                # perform retries to circumvent "ssl.SSLError: [SSL] PEM lib (_ssl.c:4012)"52                def _do_create():53                    SSL_CONTEXTS[key] = create_ssl_context_orig(self, *args, **kwargs)54                retry(_do_create, retries=3, sleep=0.5)55            return SSL_CONTEXTS[key]56    create_ssl_context_orig = Config.create_ssl_context57    Config.create_ssl_context = create_ssl_context58    # avoid "h11._util.LocalProtocolError: Too little data for declared Content-Length" for certain status codes59    def suppress_body(method, status_code):60        if status_code == 412:61            return False62        return suppress_body_orig(method, status_code)63    suppress_body_orig = hypercorn_utils.suppress_body64    hypercorn_utils.suppress_body = suppress_body65    http_stream.suppress_body = suppress_body66class HTTPErrorResponse(Exception):67    def __init__(self, *args, code=None, **kwargs):68        super(HTTPErrorResponse, self).__init__(*args, **kwargs)69        self.code = code70def run_server(port, handler=None, asynchronous=True, ssl_creds=None):71    ensure_event_loop()72    app = Quart(__name__)73    app.config['MAX_CONTENT_LENGTH'] = 256 * 1024 * 1024  # 256 MB request payload limit74    @app.route('/', methods=HTTP_METHODS, defaults={'path': ''})75    @app.route('/<path:path>', methods=HTTP_METHODS)76    async def index(path=None):77        response = await make_response('{}')78        if handler:79            data = await request.get_data()80            try:81                result = await run_sync(handler, request, data)82                if isinstance(result, Exception):83                    raise result84            except Exception as e:85                LOG.warning('Error in proxy handler for request %s %s: %s %s' %86                    (request.method, request.url, e, traceback.format_exc()))87                response.status_code = 50088                if isinstance(e, HTTPErrorResponse):89                    response.status_code = e.code or response.status_code90                return response91            if result is not None:92                is_chunked = uses_chunked_encoding(result)93                result_content = result.content or ''94                response = await make_response(result_content)95                response.status_code = result.status_code96                if is_chunked:97                    response.headers.pop('Content-Length', None)98                result.headers.pop('Server', None)99                result.headers.pop('Date', None)100                headers = {k: str(v).replace('\n', r'\n') for k, v in result.headers.items()}101                response.headers.update(headers)102                # set multi-value headers103                multi_value_headers = getattr(result, 'multi_value_headers', {})104                for key, values in multi_value_headers.items():105                    for value in values:106                        response.headers.add_header(key, value)107                # set default headers, if required108                if not is_chunked and request.method not in ['OPTIONS', 'HEAD']:109                    response_data = await response.get_data()110                    response.headers['Content-Length'] = str(len(response_data or ''))111                if 'Connection' not in response.headers:112                    response.headers['Connection'] = 'close'113        return response114    def run_app_sync(*args, loop=None, shutdown_event=None):115        kwargs = {}116        config = Config()117        cert_file_name, key_file_name = ssl_creds or (None, None)118        if cert_file_name:119            kwargs['certfile'] = cert_file_name120            config.certfile = cert_file_name121        if key_file_name:122            kwargs['keyfile'] = key_file_name123            config.keyfile = key_file_name124        setup_quart_logging()125        config.bind = ['0.0.0.0:%s' % port]126        loop = loop or ensure_event_loop()127        run_kwargs = {}128        if shutdown_event:129            run_kwargs['shutdown_trigger'] = shutdown_event.wait130        try:131            try:132                return loop.run_until_complete(serve(app, config, **run_kwargs))133            except Exception as e:134                LOG.info('Error running server event loop on port %s: %s %s' % (port, e, traceback.format_exc()))135                if 'SSL' in str(e):136                    c_exists = os.path.exists(cert_file_name)137                    k_exists = os.path.exists(key_file_name)138                    c_size = len(load_file(cert_file_name)) if c_exists else 0...

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.