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

How to use external_service_url method in localstack

Best Python code snippet using localstack_python

utils.py

Source:utils.py

1import logging2import requests3import requests_cache4from requests.exceptions import Timeout, ConnectionError5from requests.adapters import HTTPAdapter6from starlette.status import (7    HTTP_200_OK,8    HTTP_201_CREATED,9    HTTP_400_BAD_REQUEST10)11from .config import settings12def log_config():13    logging.basicConfig(14        filename='prescription.log',15        format='%(asctime)s.%(msecs)03d %(levelname)s %(module)s - %(funcName)s: %(message)s',16        datefmt='%Y-%m-%d %H:%M:%S',17        force=True18    )19    logger = logging.getLogger('prescription')20    logger.setLevel(level=logging.INFO)21    h = logging.StreamHandler()22    logger.addHandler(h)23    return logger24logger = log_config()25class Metric:26    def __init__(self, prescription):27        self.prescription = prescription28        self.metrics_data = {}29    def process(self):30        metrics_data = self._set_metrics_data()31        if metrics_data is not True:32            return metrics_data33        metrics_sent = self._send_metrics()34        return metrics_sent35    def _set_physician(self):36        requests_cache.install_cache('physicians_cache', backend='sqlite', expire_after=172800)37        session = requests.Session()38        session.mount(f'{settings.external_service_url}/physicians/', HTTPAdapter(max_retries=2))39        try:40            response = session.get(41                f'{settings.external_service_url}/physicians/{self.prescription.physician}',42                headers={'Authorization': settings.token_physicians},43                timeout=444            )45        except (ConnectionError, Timeout) as err:46            logger.error({'status_code': HTTP_400_BAD_REQUEST, 'detail': err})47            return {'error': True, 'status_code': HTTP_400_BAD_REQUEST, 'detail': err}48        if response.status_code != HTTP_200_OK:49            logger.error({'status_code': response.status_code, 'detail': response.text})50            return {'error': True, 'status_code': response.status_code, 'detail': response.text}51        physician = response.json()52        self.metrics_data['physician_id'] = int(physician.get('id'))53        self.metrics_data['physician_name'] = physician.get('name')54        self.metrics_data['physician_crm'] = physician.get('crm')55        logger.info({'status_code': response.status_code, 'detail': response.text})56        return True57    def _set_clinic(self):58        requests_cache.install_cache('clinics_cache', backend='sqlite', expire_after=259200)59        session = requests.Session()60        session.mount(f'{settings.external_service_url}/clinics/', HTTPAdapter(max_retries=3))61        try:62            response = session.get(63                f'{settings.external_service_url}/clinics/{self.prescription.clinic}',64                headers={'Authorization': settings.token_clinics},65                timeout=566            )67        except (ConnectionError, Timeout) as err:68            logger.error({'status_code': HTTP_400_BAD_REQUEST, 'detail': err})69            return False70        if response.status_code != HTTP_200_OK:71            logger.error({'status_code': response.status_code, 'detail': response.text})72            return False73        clinic = response.json()74        self.metrics_data['clinic_id'] = int(clinic.get('id'))75        self.metrics_data['clinic_name'] = clinic.get('name')76        logger.info({'status_code': response.status_code, 'detail': response.text})77        return True78    def _set_patient(self):79        requests_cache.install_cache('patients_cache', backend='sqlite', expire_after=43200)80        session = requests.Session()81        session.mount(f'{settings.external_service_url}/patients/', HTTPAdapter(max_retries=2))82        try:83            response = session.get(84                f'{settings.external_service_url}/patients/{self.prescription.patient}',85                headers={'Authorization': settings.token_patients},86                timeout=387            )88        except (ConnectionError, Timeout) as err:89            logger.error({'status_code': HTTP_400_BAD_REQUEST, 'detail': err})90            return {'error': True, 'status_code': HTTP_400_BAD_REQUEST, 'detail': err}91        if response.status_code != HTTP_200_OK:92            logger.error({'status_code': response.status_code, 'detail': response.text})93            return {'error': True, 'status_code': response.status_code, 'detail': response.text}94        patient = response.json()95        self.metrics_data['patient_id'] = int(patient.get('id'))96        self.metrics_data['patient_name'] = patient.get('name')97        self.metrics_data['patient_email'] = patient.get('email')98        self.metrics_data['patient_phone'] = patient.get('phone')99        logger.info({'status_code': response.status_code, 'detail': response.text})100        return True101    def _set_metrics_data(self):102        physician = self._set_physician()103        if physician is not True:104            return physician105        patient = self._set_patient()106        if patient is not True:107            return patient108        self._set_clinic()109        return True110    def _send_metrics(self):111        session = requests.Session()112        session.mount(f'{settings.external_service_url}/metrics/', HTTPAdapter(max_retries=5))113        try:114            response = session.post(115                f'{settings.external_service_url}/metrics',116                headers={'Authorization': settings.token_metrics, 'Content-Type': 'application/json'},117                json=self.metrics_data,118                timeout=6119            )120        except (ConnectionError, Timeout) as err:121            logger.error({'status_code': HTTP_400_BAD_REQUEST, 'detail': err})122            return {'error': True, 'status_code': HTTP_400_BAD_REQUEST, 'detail': err}123        if response.status_code != HTTP_201_CREATED:124            logger.error({'status_code': response.status_code, 'detail': response.text})125            return {'error': True, 'status_code': response.status_code, 'detail': response.text}126        logger.info({'status_code': response.status_code, 'detail': response.text})...

test_my_service.py

Source:test_my_service.py

1import os2import requests3MY_SERVICE_URL = os.environ["MY_SERVICE_URL"]4EXTERNAL_SERVICE_URL = os.environ["EXTERNAL_SERVICE_URL"]5def test_my_service():6    set_external_service_response({"status": 200, "jsonBody": {"value": 3}})7    my_service_factor = 38    response = requests.get(9        f"{MY_SERVICE_URL}/multiply-external-service?factor={my_service_factor}"10    )11    assert response.status_code == 20012    assert response.json()["result"] == 913def test_my_service_handles_external_service_downtimes():14    set_external_service_response({"fault": "CONNECTION_RESET_BY_PEER"})15    my_service_factor = 316    response = requests.get(17        f"{MY_SERVICE_URL}/multiply-external-service?factor={my_service_factor}"18    )19    assert response.status_code == 20020    assert response.json()["result"] == 321def set_external_service_response(response):22    response = requests.post(23        f"{EXTERNAL_SERVICE_URL}/__admin/mappings",24        json={25            "request": {"method": "GET", "url": "/get-value"},26            "response": response,27        },28    )...

app.py

Source:app.py

1import flask2import logging3import os4import requests5app = flask.Flask(__name__)6EXTERNAL_SERVICE_URL = os.environ["EXTERNAL_SERVICE_URL"]7@app.route("/multiply-external-service")8def multiply_external_service():9    factor = int(flask.request.args.get("factor", "1"))10    return {"result": factor * _get_external_service_value()}11def _get_external_service_value():12    try:13        external_service_response = requests.get(f"{EXTERNAL_SERVICE_URL}/get-value")14        return external_service_response.json()["value"]15    except requests.exceptions.ConnectionError:16        logging.exception("Error calling external service, let's fallback to 1.")...

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.