KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_service method in localstack
Best Python code snippet using localstack_python
rasa_blueprint.py
Source:rasa_blueprint.py 
...16app_rasa = Blueprint(route_name,__name__, url_prefix='/api')17@app_rasa.route('/reset/<id>'.format(route_name), methods=['DELETE'])18def delete_by_user(id):19 try:20 ServiceHandler.get_service(ROUTE_CONFIG['USER_EXAM_TYPE_NAME']).delete_by_user_id(id)21 ServiceHandler.get_service(ROUTE_CONFIG['USER_FORM_TYPE_NAME']).delete_by_user_id(id)22 ServiceHandler.get_service(ROUTE_CONFIG['USER_HEALTH_INFO_TYPE_NAME']).delete_by_user_id(id)23 ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).delete_by_user_id(id)24 ServiceHandler.get_service(ROUTE_CONFIG['USER_POSTPARTUM_INFO_TYPE_NAME']).delete_by_user_id(id)25 ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).delete_by_user_id(id)26 ServiceHandler.get_service(ROUTE_CONFIG['USER_TRIMESTER_TYPE_NAME']).delete_by_user_id(id)27 ServiceHandler.get_service(ROUTE_CONFIG['CONVERSATIONS_TYPE_NAME']).delete_by_user_id(id)28 return response(status=status.HTTP_200_OK)29 except Exception as e:30 return response(str(e), status.HTTP_400_BAD_REQUEST) 31@app_rasa.route('/{}/send-slot-user-exam'.format(route_name), methods=['POST'])32def send_user_exam_slot():33 try:34 json_obj = request.get_json()35 user_exam = ServiceHandler.get_service(ROUTE_CONFIG['USER_EXAM_TYPE_NAME']).get(json_obj['id'])36 if not user_exam:37 raise Exception('User Exam not found!')38 obj = {39 'exam_status' : json_obj['exam_status'],40 'id' : json_obj['id']41 }42 ServiceHandler.get_service(ROUTE_CONFIG['USER_EXAM_TYPE_NAME']).update(obj)43 return response(status=status.HTTP_201_CREATED)44 except Exception as e:45 return response_text(str(e), status.HTTP_400_BAD_REQUEST)46@app_rasa.route('/{}/send-slots/<id>'.format(route_name), methods=['POST'])47def send_slots_from_Rasa(id):48 try:49 json_obj = request.get_json()50 user = ServiceHandler.get_service(ROUTE_CONFIG['USER_TYPE_NAME']).get(id)51 if not user:52 raise Exception('User not found!')53 54 obj_user_info = {55 'has_children' : False if 'has_children' not in json_obj else json_obj['has_children'],56 'has_health_plan' : False if 'has_health_plan' not in json_obj else json_obj['has_health_plan'],57 'is_planning' : False if 'is_planning' not in json_obj else json_obj['is_planning'],58 'is_pregnant' : False if 'is_pregnant' not in json_obj else json_obj['is_pregnant'],59 'is_trying' : False if 'is_trying' not in json_obj else json_obj['is_trying'],60 'is_postpartum' : False if 'is_postpartum' not in json_obj else json_obj['is_postpartum'],61 'user' : user,62 'height': None,63 'weight': None,64 'state': None65 }66 67 user_info = ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).get_by_user_id(user.id)68 if user_info:69 obj_user_info['id'] = user_info.id70 ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).update(obj_user_info)71 else:72 ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).insert(obj_user_info)73 74 if json_obj['is_pregnant']:75 obj_pregnancy_info = {76 'is_first_pregnancy' : False if 'is_first_pregnancy' not in json_obj else json_obj['is_first_pregnancy'],77 'is_planned_pregnancy' : False if 'is_planned_pregnancy' not in json_obj else json_obj['is_planned_pregnancy'],78 'is_doing_pre_natal' : False if 'is_doing_pre_natal' not in json_obj else json_obj['is_doing_pre_natal'],79 'last_menstruation_date' : None if 'last_menstruation_date' not in json_obj else json_obj['last_menstruation_date'],80 'first_ultrasound_date' : None if 'first_ultrasound_date' not in json_obj else json_obj['first_ultrasound_date'],81 'user' : user82 }83 84 user_pregnancy_info = ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).get_by_user_id(user.id)85 if user_pregnancy_info:86 obj_pregnancy_info['id'] = user_pregnancy_info.id87 ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).update(obj_pregnancy_info)88 else:89 ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).insert(obj_pregnancy_info)90 user_trimester = ServiceHandler.get_service(ROUTE_CONFIG['USER_TRIMESTER_TYPE_NAME']).get_by_user_id(user.id)91 if not user_trimester:92 last_menstruation_date = parse(obj_pregnancy_info['last_menstruation_date']).date()93 if last_menstruation_date > date.today():94 last_menstruation_date = last_menstruation_date - relativedelta(years=1)95 96 if not obj_pregnancy_info['first_ultrasound_date']:97 start_date = last_menstruation_date98 else:99 first_ultrasound_date = parse(obj_pregnancy_info['first_ultrasound_date']).date()100 if first_ultrasound_date > date.today():101 first_ultrasound_date = first_ultrasound_date - relativedelta(years=1)102 103 date_diff = first_ultrasound_date - last_menstruation_date104 start_date = first_ultrasound_date if date_diff.days > 7 else last_menstruation_date105 106 if ((date.today() - start_date).days <= 90):107 trimester = 1108 elif ((date.today() - start_date).days <= 180):109 trimester = 2110 else:111 trimester = 3112 113 obj = {114 'start_date' : start_date.strftime("%Y-%m-%dT%H:%M:%S.%f"),115 'trimester' : trimester,116 'user': user117 }118 119 ServiceHandler.get_service(ROUTE_CONFIG['USER_TRIMESTER_TYPE_NAME']).insert(obj)120 121 if json_obj['is_postpartum']:122 obj_postpartum_info = {123 'is_breastfeeding' : False if 'is_breastfeeding' not in json_obj else json_obj['is_breastfeeding'],124 'is_having_sex' : False if 'is_having_sex' not in json_obj else json_obj['is_having_sex'],125 'contraceptive_method' : None if 'contraceptive_method' not in json_obj else json_obj['contraceptive_method'], 126 'had_doctor_appointment' : False if 'had_doctor_appointment' not in json_obj else json_obj['had_doctor_appointment'],127 'had_infection' : False if 'had_infection' not in json_obj else json_obj['had_infection'],128 'infection_kind' : None if 'infection_kind' not in json_obj else json_obj['infection_kind'],129 'user' : user130 }131 user_postpartum_info = ServiceHandler.get_service(ROUTE_CONFIG['USER_POSTPARTUM_INFO_TYPE_NAME']).get_by_user_id(user.id)132 if user_postpartum_info:133 obj_postpartum_info['id'] = user_postpartum_info.id134 ServiceHandler.get_service(ROUTE_CONFIG['USER_POSTPARTUM_INFO_TYPE_NAME']).update(obj_postpartum_info)135 else:136 ServiceHandler.get_service(ROUTE_CONFIG['USER_POSTPARTUM_INFO_TYPE_NAME']).insert(obj_postpartum_info)137 138 return response(status=status.HTTP_200_OK)139 except Exception as e:140 print(str(e))141 return response_text(str(e), status.HTTP_400_BAD_REQUEST)142 return None143@app_rasa.route('/{}/send-health-slots/<id>'.format(route_name), methods=['POST'])144def send_health_slots_from_Rasa(id):145 try:146 json_obj = request.get_json()147 user = ServiceHandler.get_service(ROUTE_CONFIG['USER_TYPE_NAME']).get(id)148 if not user:149 raise Exception('User not found!')150 obj = {151 'takes_regular_medicine' : False if 'takes_regular_medicine' not in json_obj else json_obj['takes_regular_medicine'],152 'regular_medicine_name' : None if 'regular_medicine_name' not in json_obj else json_obj['regular_medicine_name'],153 'has_hypothyroidism' : False if 'has_hypothyroidism' not in json_obj else json_obj['has_hypothyroidism'],154 'has_hyperthyroidism' : False if 'has_hyperthyroidism' not in json_obj else json_obj['has_hyperthyroidism'],155 'has_diabetes' : False if 'has_diabetes' not in json_obj else json_obj['has_diabetes'],156 'drug_use' : False if 'drug_use' not in json_obj else json_obj['drug_use'],157 'has_autoimmune_disease' : False if 'has_autoimmune_disease' not in json_obj else json_obj['has_autoimmune_disease'],158 'has_asthma' : False if 'has_asthma' not in json_obj else json_obj['has_asthma'],159 'is_seropositive' : False if 'is_seropositive' not in json_obj else json_obj['is_seropositive'],160 'has_high_pressure' : False if 'has_high_pressure' not in json_obj else json_obj['has_high_pressure'],161 'user' : user162 }163 user_health_info = ServiceHandler.get_service(ROUTE_CONFIG['USER_HEALTH_INFO_TYPE_NAME']).get_by_user_id(user.id)164 165 if user_health_info:166 obj['id'] = user_health_info.id167 ServiceHandler.get_service(ROUTE_CONFIG['USER_HEALTH_INFO_TYPE_NAME']).update(obj)168 else:169 ServiceHandler.get_service(ROUTE_CONFIG['USER_HEALTH_INFO_TYPE_NAME']).insert(obj)170 171 return response(status=status.HTTP_200_OK)172 except Exception as e:173 print(str(e))174 return response_text(str(e), status.HTTP_400_BAD_REQUEST)175 return None176@app_rasa.route('/{}/send-pregnancy-slots/<id>'.format(route_name), methods=['POST'])177def send_pregnancy_slots_from_Rasa(id):178 try:179 json_obj = request.get_json()180 181 user = ServiceHandler.get_service(ROUTE_CONFIG['USER_TYPE_NAME']).get(id)182 if not user:183 raise Exception('User not found!')184 obj = {185 'current_high_risk' : False if 'current_high_risk' not in json_obj else json_obj['current_high_risk'],186 'due_date' : None if 'due_date' not in json_obj else json_obj['due_date'],187 'births' : None if 'births' not in json_obj else json_obj['births'],188 'cesarean_births' : None if 'cesarean_births' not in json_obj else json_obj['cesarean_births'],189 'normal_births' : None if 'normal_births' not in json_obj else json_obj['normal_births'],190 'why_cesarean_birth' : None if 'why_cesarean_birth' not in json_obj else json_obj['why_cesarean_birth'],191 'abortion' : False if 'abortion' not in json_obj else json_obj['abortion'],192 'premature_birth' : False if 'premature_birth' not in json_obj else json_obj['premature_birth'],193 'user' : user194 }195 196 user_pregnancy_info = ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).get_by_user_id(user.id)197 198 if user_pregnancy_info:199 obj['id'] = user_pregnancy_info.id200 ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).update(obj)201 else:202 ServiceHandler.get_service(ROUTE_CONFIG['USER_PREGNANCY_INFO_TYPE_NAME']).insert(obj)203 204 return response(status=status.HTTP_200_OK)205 except Exception as e:206 print(str(e))207 return response_text(str(e), status.HTTP_400_BAD_REQUEST)208 return None209@app_rasa.route('/{}/send-personal-slots/<id>'.format(route_name), methods=['POST'])210def send_personal_slots_from_Rasa(id):211 try:212 json_obj = request.get_json()213 user = ServiceHandler.get_service(ROUTE_CONFIG['USER_TYPE_NAME']).get(id)214 if not user:215 raise Exception('User not found!')216 obj = {217 'height' : None if 'height' not in json_obj else json_obj['height'],218 'weight' : None if 'weight' not in json_obj else json_obj['weight'],219 'state' : None if 'state' not in json_obj else json_obj['state'],220 'user' : user221 }222 223 user_info = ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).get_by_user_id(user.id)224 225 if user_info:226 obj['id'] = user_info.id227 ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).update(obj)228 else:229 obj_info = {230 'has_children': None,231 'has_health_plan': None,232 'is_planning': None,233 'is_pregnant': None,234 'is_trying': None,235 'is_postpartum': None,236 }237 obj.update(obj_info)238 ServiceHandler.get_service(ROUTE_CONFIG['USER_INFO_TYPE_NAME']).insert(obj)239 240 return response(status=status.HTTP_200_OK)241 except Exception as e:242 print(str(e))243 return response_text(str(e), status.HTTP_400_BAD_REQUEST)...
routes.py
Source:routes.py
...10 return Response(status=204)11#############################################################12@api.route("Document/<int:houseId>/Tenant")13def get_tenant_documents(houseId):14 tenantService = zookeeper.get_service("tenant-service")15 if not tenantService:16 return Response(response="Error: Tenant Service Currently Unavailable", status=503)17 tenantManager = RequestManager(request, tenantService)18 tenantId = tenantManager.authenticate()19 if not tenantId:20 return Response(response="Not Authorized", status=401)21 documentService = zookeeper.get_service("document-service")22 if not documentService:23 return Response(response="Error: Document Service Currently Unavailable", status=503)24 documentManager = RequestManager(request, documentService)25 return documentManager.get("document/v1/Document/" + str(houseId) + "/Tenant")26 27 28@api.route("House/<int:houseId>")29def get_house(houseId):30 tenantService = zookeeper.get_service("tenant-service")31 if not tenantService:32 return Response(response="Error: Tenant Service Currently Unavailable", status=503)33 tenantManager = RequestManager(request, tenantService)34 tenantId = tenantManager.authenticate()35 if not tenantId:36 return Response(response="Not Authorized", status=401)37 houseService = zookeeper.get_service("house-service")38 if not houseService:39 return Response(response="Error: House Service Currently Unavailable", status=503)40 houseManager = RequestManager(request, houseService)41 return houseManager.get("house/v1/House/" + str(houseId) + "/Tenant")42 43 44@api.route("/RentDetails/<int:houseId>")45def get_rent_details(houseId):46 tenantService = zookeeper.get_service("tenant-service")47 if not tenantService:48 return Response(response="Error: Tenant Service Currently Unavailable", status=503)49 tenantManager = RequestManager(request, tenantService)50 tenantId = tenantManager.authenticate()51 if not tenantId:52 return Response(response="Not Authorized", status=401)53 leaseService = zookeeper.get_service("lease-service")54 if not leaseService:55 return Response(response="Error: Rent Service Currently Unavailable", status=503)56 leaseManager = RequestManager(request, leaseService)57 return leaseManager.get("lease/v1/RentDetails/" + str(houseId) + "/Tenant")58 59 60@api.route("/Homeowner/<int:houseId>")61def get_homeowner(houseId):62 tenantService = zookeeper.get_service("tenant-service")63 if not tenantService:64 return Response(response="Error: Tenant Service Currently Unavailable", status=503)65 tenantManager = RequestManager(request, tenantService)66 tenantId = tenantManager.authenticate()67 if not tenantId:68 return Response(response="Not Authorized", status=401)69 houseService = zookeeper.get_service("house-service")70 if not houseService:71 return Response(response="Error: House Service Currently Unavailable", status=503)72 houseManager = RequestManager(request, houseService)73 houseResponse = houseManager.get("house/v1/House/" + str(houseId) + "/Tenant")74 houseData = houseResponse.get_json()75 homeownerId = houseData["homeownerId"]76 if not homeownerId:77 return Response(response="Error: Homeowner Not Found", status=404)78 homeownerService = zookeeper.get_service("homeowner-service")79 if not homeownerService:80 return Response(response="Error: Homeowner Service Currently Unavailable", status=503)81 homeownerManager = RequestManager(request, homeownerService)82 return homeownerManager.get("homeowner/v1/Homeowner/" + str(homeownerId))83 84 85 86#############################################################87@api.route("/")88def get_tenant_account():89 service = zookeeper.get_service("tenant-service")90 if not service:91 return Response(response="Error: Tenant Service Currently Unavailable", status=503)92 manager = RequestManager(request, service)93 return manager.get_html("/tenant/v1/")94 95@api.route("/", methods=["POST"])96def create_tenant_account():97 service = zookeeper.get_service("tenant-service")98 if not service:99 return Response(response="Error: Tenant Service Currently Unavailable", status=503)100 manager = RequestManager(request, service)101 return manager.post_html("/tenant/v1/")102 103@api.route("Tenant", methods=["GET"])104def get_tenant():105 tenantService = zookeeper.get_service("tenant-service")106 if not tenantService:107 return Response(response="Error: Homeowner Not Available", status=503)108 tenantManager = RequestManager(request, tenantService)109 tenantId = tenantManager.authenticate()110 if not tenantId:111 return Response(response="Not Authorized", status=401)112 return tenantManager.get("tenant/v1/Tenant")113 114##########################################################115@api.route("Login", methods=["POST"])116def login_tenant():117 service = zookeeper.get_service("tenant-service")118 if not service:119 return Response(response="Error: Zookeeper down", status=503)120 manager = RequestManager(request, service)121 return manager.post("tenant/v1/Login")122 123 124####################################################125@api.route("Problem", methods=["POST"])126def create_problem():127 tenantService = zookeeper.get_service("tenant-service")128 if not tenantService:129 return Response(response="Error: Tenant Service Currently Unavailable", status=503)130 tenantManager = RequestManager(request, tenantService)131 tenantId = tenantManager.authenticate()132 if not tenantId:133 return Response(response="Not Authorized", status=401)134 problemService = zookeeper.get_service("problem-service")135 if not problemService:136 return Response(response="Error: Problem Service Currently Unavailable", status=503)137 problemManager = RequestManager(request, problemService)138 return problemManager.post_problem()139 140 141 142@api.route("House/<int:houseId>/Problem")143def get_problems(houseId):144 tenantService = zookeeper.get_service("tenant-service")145 if not tenantService:146 return Response(response="Error: Tenant Service Currently Unavailable", status=503)147 tenantManager = RequestManager(request, tenantService)148 tenantId = tenantManager.authenticate()149 if not tenantId:150 return Response(response="Not Authorized", status=401)151 problemService = zookeeper.get_service("problem-service")152 if not problemService:153 return Response(response="Error: Problem Service Currently Unavailable", status=503)154 problemManager = RequestManager(request, problemService)155 return problemManager.get("problem/v1/House/" + str(houseId) + "/Problem")156 157 158@api.route("Tenant/Profile", methods=["POST"])159def tenantProfile():160 tenantService = zookeeper.get_service("tenant-service")161 if not tenantService:162 return Response(response="Error: Homeowner Not Available", status=503)163 tenantManager = RequestManager(request, tenantService)164 tenantId = tenantManager.authenticate()165 if not tenantId:166 return Response(response="Not Authorized", status=401)167 imageUploadService = zookeeper.get_service("image-upload-service")168 if not imageUploadService:169 return Response(response="Error: Image Upload Not Available", status=503)170 imageUploadManager = RequestManager(request, imageUploadService)171 return imageUploadManager.post_profile(tenantId)172 173 174 ...
test_plan.py
Source:test_plan.py
...17 ("agent",""),18 ("thing",""),19 ("destination","")]20 for call in calls:21 rospy.loginfo(json.loads(get_service(*call).response))22 # Try Adding an Agent23 rospy.loginfo("Creating an Agent.")24 response = create_service("agent","{}")25 if not response.success:26 rospy.loginfo(response.message)27 response = json.loads(get_service("agent","").response)28 rospy.loginfo(response)29 agent_id = response.keys()[0]30 response = json.loads(get_service("agent",agent_id).response)31 rospy.loginfo(response)32 # Try adding a Therblig33 rospy.loginfo("Creating a Therblig.")34 tasks = json.loads(get_service("task_keys","").response)35 parameters = json.dumps({"index":0,"task_id":tasks[0],"type":"transport_empty"})36 response = create_service("therblig",parameters)37 if not response.success:38 rospy.loginfo(response.message)39 response = json.loads(get_service("therblig","").response)40 rospy.loginfo(response)41 # Try adding a Thing42 rospy.loginfo("Creating a Thing.")43 response = create_service("thing","{}")44 if not response.success:45 rospy.loginfo(response.message)46 response = json.loads(get_service("thing","").response)47 rospy.loginfo(response)48 response = json.loads(get_service("destination","").response)49 rospy.loginfo(response)50 # Try adding a Destination51 rospy.loginfo("Creating a Destination.")52 response = create_service("destination","{}")53 if not response.success:54 rospy.loginfo(response.message)55 response = json.loads(get_service("destination","").response)56 rospy.loginfo(response)57 rospy.loginfo("Deleting the Therblig")58 therbligs = json.loads(get_service("therblig_keys","").response)59 rospy.loginfo("therbligs before:")60 rospy.loginfo(therbligs)61 response = delete_service("therblig", therbligs[0])62 if not response.success:63 rospy.loginfo("Failed to delete the therblig")64 else:65 rospy.loginfo("Therblig successfully deleted")66 therbligs = json.loads(get_service("therblig_keys","").response)67 rospy.loginfo("therbligs after:")68 rospy.loginfo(therbligs)69 rospy.loginfo("Deleting the Task")70 tasks = json.loads(get_service("task_keys","").response)71 rospy.loginfo("tasks before:")72 rospy.loginfo(tasks)73 response = delete_service("task", tasks[0])74 if not response.success:75 rospy.loginfo("Failed to delete the task")76 else:77 rospy.loginfo("Task successfully deleted")78 tasks = json.loads(get_service("task_keys","").response)79 rospy.loginfo("tasks after:")80 rospy.loginfo(tasks)81if __name__ == '__main__':82 try:83 tester()84 except rospy.ROSInterruptException:...
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?
