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

How to use _forward_request method in localstack

Best Python code snippet using localstack_python

enrichment_visualisation.py

Source:enrichment_visualisation.py

...9from neo4japp.exceptions import StatisticalEnrichmentError10from requests.exceptions import ConnectionError11bp = Blueprint('enrichment-visualisation-api', __name__, url_prefix='/enrichment-visualisation')12SE_URL = os.getenv('STATISTICAL_ENRICHMENT_URL', 'http://statistical-enrichment:5000')13def _forward_request(resource):14    url = f'{SE_URL}{resource}'15    try:16        resp = requests.request(17                method=request.method,18                url=url,19                headers={key: value for (key, value) in request.headers if key != 'Host'},20                data=request.get_data(),21                cookies=request.cookies,22                allow_redirects=False,23                timeout=120,24        )25        excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']26        headers = [27            (name, value) for (name, value) in resp.raw.headers.items()28            if name.lower() not in excluded_headers29        ]30    except ConnectionError as e:31        raise StatisticalEnrichmentError(32            'Unable to process request',33            'An unexpected connection error occurred to statistical enrichment service.'34        )35    # 500 should contain message from service so we try to include it36    if resp.status_code == 500:37        try:38            decoded_error_message = json.loads(resp.content)['message']39        except Exception as e:40            # log and proceed so general error can be raised41            logging.exception(e)42        else:43            raise StatisticalEnrichmentError(44                    'Statistical enrichment error',45                    decoded_error_message,46                    code=500,47            )48    # All errors including failure to parse internal error message49    if 400 <= resp.status_code < 600:50        raise StatisticalEnrichmentError(51                'Unable to process request',52                'An error of statistical enrichment service occurred.',53                code=resp.status_code54        )55    return Response(resp.content, resp.status_code, headers)56@bp.route('/enrich-with-go-terms', methods=['POST'])57@auth.login_required58def enrich_go():...

views.py

Source:views.py

...4from django.shortcuts import get_object_or_4045from proxy_api.callbacks.models import Callback, CallbackCallLog6from proxy_api.core.utils import extract_headers, get_response_data_for_request, \7    get_path8def _forward_request(callback, path, headers, body, method):9    callback_request = urllib.request.Request(10        callback.url + "/" + path, data=body,11        headers=headers, method=method)12    response_kwargs = get_response_data_for_request(callback_request)13    CallbackCallLog.objects.create(callback=callback,14                                   request_path=path,15                                   request_body=body,16                                   request_body_binary=body,17                                   request_headers={name: str(value)18                                                    for name, value19                                                    in headers.items()},20                                   **response_kwargs)21    return HttpResponse(**response_kwargs)22def forward(request, callback_slug, callback_path=""):23    callback = get_object_or_404(Callback, slug=callback_slug)24    if not callback.methods.filter(name=request.method).exists():25        return HttpResponse(26            "Method {} for API '{}' is not allowed".format(27                request.method, callback_slug), status=405)28    path = get_path(callback_path, request)29    callback_headers = extract_headers(request.META)30    return _forward_request(callback, path, callback_headers,31                            request.body, request.method)32def replay(request, id):33    callback_call_log = get_object_or_404(CallbackCallLog, id=id)34    callback = callback_call_log.callback35    callback_path = callback_call_log.request_path36    callback_headers = extract_headers(callback_call_log.request_headers)37    callback_method = callback_call_log.request_headers['REQUEST_METHOD']38    callback_body = bytes(callback_call_log.request_body_binary)39    return _forward_request(callback, callback_path, callback_headers,...

sep.py

Source:sep.py

...11        self.protocol = self._protocol_class()12        self.reset()13    def reset(self):14        self.buffer = b''15    def _forward_request(self, request, idx):16        assert idx > 017        data = self.buffer[:idx]18        self.protocol.event(_request(self, parent=request, data=data))19        self.buffer = self.buffer[idx + len(self.separator):]20        return21    def event(self, request):22        from fenestre.network import events23        if request.event is events.receive:24            self.buffer += request.data25            if len(self.buffer) >= self.max_size:26                dsize = len(self.buffer) - self.max_size27                self.buffer = self.buffer[dsize:]28            while True:29                idx = self.buffer.find(self.separator)30                if idx > 0:31                    self._forward_request(request, idx)32                else:33                    return34        else:...

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.