Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
