Best Python code snippet using localstack_python
slices_controller.py
Source:slices_controller.py
...72 return cors_success_response(response_body=response)73 except OrchestratorException as e:74 logger.exception(e)75 failure_counter.labels(POST_METHOD, SLICES_CREATE_PATH).inc()76 return cors_error_response(error=e)77 except Exception as e:78 logger.exception(e)79 failure_counter.labels(POST_METHOD, SLICES_CREATE_PATH).inc()80 return cors_error_response(error=e)81def slices_delete_slice_id_delete(slice_id) -> Status200OkNoContent: # noqa: E50182 """Delete slice.83 Request to delete slice. On success, resources associated with slice or sliver are stopped if necessary,84 de-provisioned and un-allocated at the respective sites. # noqa: E50185 :param slice_id: Slice identified by universally unique identifier86 :type slice_id: str87 :rtype: Status200OkNoContent88 """89 handler = OrchestratorHandler()90 logger = handler.get_logger()91 received_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()92 try:93 token = get_token()94 handler.delete_slice(token=token, slice_id=slice_id)95 success_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()96 slice_info = Status200OkNoContentData()97 slice_info.details = f"Slice '{slice_id}' has been successfully deleted"98 response = Status200OkNoContent()99 response.data = [slice_info]100 response.size = len(response.data)101 response.status = 200102 response.type = 'no_content'103 return cors_success_response(response_body=response)104 except OrchestratorException as e:105 logger.exception(e)106 failure_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()107 return cors_error_response(error=e)108 except Exception as e:109 logger.exception(e)110 failure_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()111 return cors_error_response(error=e)112def slices_get(name=None, states=None, limit=None, offset=None) -> Slices: # noqa: E501113 """Retrieve a listing of user slices114 Retrieve a listing of user slices # noqa: E501115 :param name: Search for Slices with the name116 :type name: str117 :param states: Search for Slices in the specified states118 :type states: List[str]119 :param limit: maximum number of results to return per page (1 or more)120 :type limit: int121 :param offset: number of items to skip before starting to collect the result set122 :type offset: int123 :rtype: Slices124 """125 handler = OrchestratorHandler()126 logger = handler.get_logger()127 received_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()128 try:129 token = get_token()130 slices_dict = handler.get_slices(token=token, states=states, name=name, limit=limit, offset=offset)131 response = Slices()132 response.data = []133 response.type = 'slices'134 for s in slices_dict:135 slice_obj = Slice().from_dict(s)136 response.data.append(slice_obj)137 response.size = len(response.data)138 success_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()139 return cors_success_response(response_body=response)140 except OrchestratorException as e:141 logger.exception(e)142 failure_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()143 return cors_error_response(error=e)144 except Exception as e:145 logger.exception(e)146 failure_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()147 return cors_error_response(error=e)148def slices_modify_slice_id_accept_post(slice_id): # noqa: E501149 """Accept the last modify an existing slice150 Accept the last modify and prune any failed resources from the Slice.151 Also return the accepted slice model back to the user. # noqa: E501152 :param slice_id: Slice identified by universally unique identifier153 :type slice_id: str154 :rtype: Slivers155 """156 handler = OrchestratorHandler()157 logger = handler.get_logger()158 received_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()159 try:160 token = get_token()161 value = handler.modify_accept(token=token, slice_id=slice_id)162 slice_object = Slice().from_dict(value)163 response = SliceDetails(data=[slice_object], size=1)164 response.type = 'slice_details'165 success_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()166 return cors_success_response(response_body=response)167 except OrchestratorException as e:168 logger.exception(e)169 failure_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()170 return cors_error_response(error=e)171 except Exception as e:172 logger.exception(e)173 failure_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()174 return cors_error_response(error=e)175def slices_modify_slice_id_put(body, slice_id): # noqa: E501176 """Modify an existing slice177 Request to modify an existing slice as described in the request. Request would be a graph ML describing the178 experiment topolgy expected after a modify. The supported modify actions include adding or removing nodes,179 components, network services or interfaces of the slice. On success, one or more slivers are allocated,180 containing resources satisfying the request, and assigned to the given slice. This API returns list and181 description of the resources reserved for the slice in the form of Graph ML. Orchestrator would also trigger182 provisioning of these resources asynchronously on the appropriate sites either now or in the future as requested.183 Experimenter can invoke get slice API to get the latest state of the requested resources. # noqa: E501184 :param body:185 :type body: dict | bytes186 :param slice_id: Slice identified by universally unique identifier187 :type slice_id: str188 :rtype: Slivers189 """190 handler = OrchestratorHandler()191 logger = handler.get_logger()192 received_counter.labels(POST_METHOD, SLICES_MODIFY_PATH).inc()193 try:194 token = get_token()195 slice_graph = body.decode("utf-8")196 slivers_dict = handler.modify_slice(token=token, slice_id=slice_id, slice_graph=slice_graph)197 response = Slivers()198 response.data = []199 for s in slivers_dict:200 sliver = Sliver().from_dict(s)201 response.data.append(sliver)202 response.size = len(response.data)203 response.type = "slivers"204 success_counter.labels(POST_METHOD, SLICES_MODIFY_PATH).inc()205 return cors_success_response(response_body=response)206 except OrchestratorException as e:207 logger.exception(e)208 failure_counter.labels(POST_METHOD, SLICES_MODIFY_PATH).inc()209 return cors_error_response(error=e)210 except Exception as e:211 logger.exception(e)212 failure_counter.labels(POST_METHOD, SLICES_CREATE_PATH).inc()213 return cors_error_response(error=e)214def slices_renew_slice_id_post(slice_id, lease_end_time) -> Status200OkNoContent: # noqa: E501215 """Renew slice216 Request to extend slice be renewed with their expiration extended. If possible, the orchestrator should extend the217 slivers to the requested expiration time, or to a sooner time if policy limits apply. # noqa: E501218 :param slice_id: Slice identified by universally unique identifier219 :type slice_id: str220 :param lease_end_time: New Lease End Time for the Slice221 :type lease_end_time: str222 :rtype: Status200OkNoContent223 """224 handler = OrchestratorHandler()225 logger = handler.get_logger()226 received_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()227 try:228 token = get_token()229 handler.renew_slice(token=token, slice_id=slice_id, new_lease_end_time=lease_end_time)230 success_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()231 slice_info = Status200OkNoContentData()232 slice_info.details = f"Slice '{slice_id}' has been successfully renewed"233 response = Status200OkNoContent()234 response.data = [slice_info]235 response.size = len(response.data)236 response.status = 200237 response.type = 'no_content'238 return cors_success_response(response_body=response)239 except OrchestratorException as e:240 logger.exception(e)241 failure_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()242 return cors_error_response(error=e)243 except Exception as e:244 logger.exception(e)245 failure_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()246 return cors_error_response(error=e)247def slices_slice_id_get(slice_id, graph_format) -> SliceDetails: # noqa: E501248 """slice properties249 Retrieve Slice properties # noqa: E501250 :param slice_id: Slice identified by universally unique identifier251 :type slice_id: str252 :param graph_format: graph format253 :type graph_format: str254 :rtype: SliceDetails255 """256 handler = OrchestratorHandler()257 logger = handler.get_logger()258 received_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()259 try:260 token = get_token()261 value = handler.get_slice_graph(token=token, slice_id=slice_id, graph_format_str=graph_format)262 slice_object = Slice().from_dict(value)263 response = SliceDetails(data=[slice_object], size=1)264 response.type = 'slice_details'265 success_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()266 return cors_success_response(response_body=response)267 except OrchestratorException as e:268 logger.exception(e)269 failure_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()270 return cors_error_response(error=e)271 except Exception as e:272 logger.exception(e)273 failure_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()...
slivers_controller.py
Source:slivers_controller.py
...55 return cors_success_response(response_body=response)56 except OrchestratorException as e:57 logger.exception(e)58 failure_counter.labels(GET_METHOD, SLIVERS_GET_PATH).inc()59 return cors_error_response(error=e)60 except Exception as e:61 logger.exception(e)62 failure_counter.labels(GET_METHOD, SLIVERS_GET_PATH).inc()63 return cors_error_response(error=e)64def slivers_sliver_id_get(slice_id, sliver_id) -> Slivers: # noqa: E50165 """slivers properties66 Retrieve Sliver properties # noqa: E50167 :param slice_id: Slice identified by universally unique identifier68 :type slice_id: str69 :param sliver_id: Sliver identified by universally unique identifier70 :type sliver_id: str71 :rtype: Slivers72 """73 handler = OrchestratorHandler()74 logger = handler.get_logger()75 received_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()76 try:77 token = get_token()78 slivers_dict = handler.get_slivers(slice_id=slice_id, token=token, sliver_id=sliver_id)79 response = Slivers()80 response.data = []81 for s in slivers_dict:82 sliver = Sliver().from_dict(s)83 response.data.append(sliver)84 response.size = len(response.data)85 response.type = "slivers"86 success_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()87 return cors_success_response(response_body=response)88 except OrchestratorException as e:89 logger.exception(e)90 failure_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()91 return cors_error_response(error=e)92 except Exception as e:93 logger.exception(e)94 failure_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()...
resources_controller.py
Source:resources_controller.py
...50 return cors_success_response(response_body=response)51 except OrchestratorException as e:52 logger.exception(e)53 failure_counter.labels(GET_METHOD, PORTAL_RESOURCES_PATH).inc()54 return cors_error_response(error=e)55 except Exception as e:56 logger.exception(e)57 failure_counter.labels(GET_METHOD, PORTAL_RESOURCES_PATH).inc()58 return cors_error_response(error=e)59def resources_get(level, force_refresh) -> Resources: # noqa: E50160 """Retrieve a listing and description of available resources61 Retrieve a listing and description of available resources # noqa: E50162 :param level: Level of details63 :type level: int64 :param force_refresh: Force to retrieve current available resource information.65 :type force_refresh: bool66 :rtype: Resources67 """68 handler = OrchestratorHandler()69 logger = handler.get_logger()70 received_counter.labels(GET_METHOD, RESOURCES_PATH).inc()71 try:72 token = get_token()73 bqm_dict = handler.list_resources(token=token, level=level, force_refresh=force_refresh)74 response = Resources()75 response.data = [Resource().from_dict(bqm_dict)]76 response.size = 177 response.type = "resources"78 success_counter.labels(GET_METHOD, RESOURCES_PATH).inc()79 return cors_success_response(response_body=response)80 except OrchestratorException as e:81 logger.exception(e)82 failure_counter.labels(GET_METHOD, RESOURCES_PATH).inc()83 return cors_error_response(error=e)84 except Exception as e:85 logger.exception(e)86 failure_counter.labels(GET_METHOD, RESOURCES_PATH).inc()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!