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

How to use _create_endpoint method in localstack

Best Python code snippet using localstack_python

transaction.py

Source:transaction.py

...44    )45    schema_href: Optional[str] = attr.ib(default=None)46    router: APIRouter = attr.ib(factory=APIRouter)47    response_class: Type[Response] = attr.ib(default=JSONResponse)48    def _create_endpoint(49        self,50        func: Callable,51        request_type: Union[52            Type[APIRequest],53            Type[BaseModel],54            Type[stac_types.Item],55            Type[stac_types.Collection],56        ],57    ) -> Callable:58        """Create a FastAPI endpoint."""59        if isinstance(self.client, AsyncBaseTransactionsClient):60            return create_async_endpoint(61                func, request_type, response_class=self.response_class62            )63        elif isinstance(self.client, BaseTransactionsClient):64            return create_sync_endpoint(65                func, request_type, response_class=self.response_class66            )67        raise NotImplementedError68    def register_create_item(self):69        """Register create item endpoint (POST /collections/{collection_id}/items)."""70        self.router.add_api_route(71            name="Create Item",72            path="/collections/{collection_id}/items",73            response_model=Item if self.settings.enable_response_models else None,74            response_class=self.response_class,75            response_model_exclude_unset=True,76            response_model_exclude_none=True,77            methods=["POST"],78            endpoint=self._create_endpoint(self.client.create_item, PostItem),79        )80    def register_update_item(self):81        """Register update item endpoint (PUT /collections/{collection_id}/items)."""82        self.router.add_api_route(83            name="Update Item",84            path="/collections/{collection_id}/items/{item_id}",85            response_model=Item if self.settings.enable_response_models else None,86            response_class=self.response_class,87            response_model_exclude_unset=True,88            response_model_exclude_none=True,89            methods=["PUT"],90            endpoint=self._create_endpoint(self.client.update_item, PutItem),91        )92    def register_delete_item(self):93        """Register delete item endpoint (DELETE /collections/{collection_id}/items/{item_id})."""94        self.router.add_api_route(95            name="Delete Item",96            path="/collections/{collection_id}/items/{item_id}",97            response_model=Item if self.settings.enable_response_models else None,98            response_class=self.response_class,99            response_model_exclude_unset=True,100            response_model_exclude_none=True,101            methods=["DELETE"],102            endpoint=self._create_endpoint(self.client.delete_item, ItemUri),103        )104    def register_create_collection(self):105        """Register create collection endpoint (POST /collections)."""106        self.router.add_api_route(107            name="Create Collection",108            path="/collections",109            response_model=Collection if self.settings.enable_response_models else None,110            response_class=self.response_class,111            response_model_exclude_unset=True,112            response_model_exclude_none=True,113            methods=["POST"],114            endpoint=self._create_endpoint(115                self.client.create_collection, stac_types.Collection116            ),117        )118    def register_update_collection(self):119        """Register update collection endpoint (PUT /collections)."""120        self.router.add_api_route(121            name="Update Collection",122            path="/collections",123            response_model=Collection if self.settings.enable_response_models else None,124            response_class=self.response_class,125            response_model_exclude_unset=True,126            response_model_exclude_none=True,127            methods=["PUT"],128            endpoint=self._create_endpoint(129                self.client.update_collection, stac_types.Collection130            ),131        )132    def register_delete_collection(self):133        """Register delete collection endpoint (DELETE /collections/{collection_id})."""134        self.router.add_api_route(135            name="Delete Collection",136            path="/collections/{collection_id}",137            response_model=Collection if self.settings.enable_response_models else None,138            response_class=self.response_class,139            response_model_exclude_unset=True,140            response_model_exclude_none=True,141            methods=["DELETE"],142            endpoint=self._create_endpoint(143                self.client.delete_collection, CollectionUri144            ),145        )146    def register(self, app: FastAPI) -> None:147        """Register the extension with a FastAPI application.148        Args:149            app: target FastAPI application.150        Returns:151            None152        """153        self.router.prefix = app.state.router_prefix154        self.register_create_item()155        self.register_update_item()156        self.register_delete_item()...

asset_search.py

Source:asset_search.py

...43    )44    asset_search_post_request_model: Type[AssetSearchPostRequest] = attr.ib(45        default=AssetSearchPostRequest46    )47    def _create_endpoint(48        self,49        func: Callable,50        request_type: Union[Type[APIRequest], Type[BaseModel],],51    ) -> Callable:52        """Create a FastAPI endpoint."""53        if isinstance(self.client, AsyncBaseAssetSearchClient):54            return create_async_endpoint(55                func, request_type, response_class=self.response_class56            )57        elif isinstance(self.client, BaseAssetSearchClient):58            return create_sync_endpoint(59                func, request_type, response_class=self.response_class60            )61        raise 62    def register_get_asset_search(self):63        """Register asset search endpoint (GET /asset/search).64        Returns:65            None66        """67        self.router.add_api_route(68            name="Asset Search",69            path="/asset/search",70            response_model=AssetCollection71            if self.settings and self.settings.enable_response_models72            else None,73            response_class=self.response_class,74            response_model_exclude_unset=True,75            response_model_exclude_none=True,76            methods=["GET"],77            endpoint=self._create_endpoint(self.client.get_asset_search, self.asset_search_get_request_model),78        )79    def register_post_asset_search(self):80        """Register asset search endpoint (POST /asset/search).81        Returns:82            None83        """84        self.router.add_api_route(85            name="Asset Search",86            path="/asset/search",87            response_model=AssetCollection88            if self.settings and self.settings.enable_response_models89            else None,90            response_class=self.response_class,91            response_model_exclude_unset=True,92            response_model_exclude_none=True,93            methods=["POST"],94            endpoint=self._create_endpoint(self.client.post_asset_search, self.asset_search_post_request_model),95        )96    97    def register_get_assets(self):98        """Register asset search endpoint (GET /collection/{collection_id}/items/{item_id}/assets).99        Returns:100            None101        """102        self.router.add_api_route(103            name="Get Assets",104            path="/collections/{collection_id}/items/{item_id}/assets",105            response_model=AssetCollection106            if self.settings and self.settings.enable_response_models107            else None,108            response_class=self.response_class,109            response_model_exclude_unset=True,110            response_model_exclude_none=True,111            methods=["GET"],112            endpoint=self._create_endpoint(self.client.get_assets, GetAssetsRequest),113        )114    115    def register_get_asset(self):116        """Register asset search endpoint (GET /collection/{collection_id}/items/{item_id}/assets/{asset_id}).117        Returns:118            None119        """120        self.router.add_api_route(121            name="Get Asset",122            path="/collections/{collection_id}/items/{item_id}/assets/{asset_id}",123            response_model=Asset124            if self.settings and self.settings.enable_response_models125            else None,126            response_class=self.response_class,127            response_model_exclude_unset=True,128            response_model_exclude_none=True,129            methods=["GET"],130            endpoint=self._create_endpoint(self.client.get_asset, GetAssetRequest),131        )132    def register(self, app: FastAPI) -> None:133        self.register_get_asset_search()134        self.register_post_asset_search()135        self.register_get_assets()136        self.register_get_asset()...

order_handler_strategy.py

Source:order_handler_strategy.py

...11    @abc.abstractmethod12    def cancel_order(self, request, *args, **kwargs):13        raise NotImplemented('Concrete implementation of method not found')14class EagerOrderHandlerStrategy(OrderHandlerStrategy):15    def _create_endpoint(self, service, method):16        base_url = service['url']17        route = service['api'][method.upper()]18        return f'{base_url}{route}'19    def place_order(self, request, *args, **kwargs):20        data = request.data21        logging.info('#'*50)22        logging.info(f'Placing order with data {data}')23        payment_data = data.pop('payment')24        # 1. Create Order25        # print(type(data))26        order = self._request.post(url=self._create_endpoint(settings.ORDER_SERVICE, 'create'),27                                   json=data)28        logging.info(f'Created order {order}')29        # 2.1. Crete payment30        payment_data['order'] = order['id']31        payment_data['order_number'] = order['order_number']32        payment_data['customer'] = order['customer']33        payment = self._request.post(url=self._create_endpoint(settings.PAYMENT_SERVICE, 'create'),34                                     json=payment_data)35        logging.info(f'Created payment {payment}')36        # 2.2 Update order37        update_data = {38            'payment': payment['id']39        }40        update_endpoint = self._create_endpoint(settings.ORDER_SERVICE, 'update')41        update_endpoint = update_endpoint.format(id=order['id'])42        order = self._request.patch(url=update_endpoint, json=update_data)43        order['payment'] = payment44        logging.info(f'Updated order {order}')45        # 3. Update Inventory46        order_items = data.pop('order_items')47        for order_item in order_items:48            order_item = {49                'product': order_item['product'],50                'quantity': order_item['quantity'],51                'price': order_item['price']52            }53            update_endpoint = self._create_endpoint(settings.PRODUCT_SERVICE, 'update')54            update_endpoint = update_endpoint.format(id=order_item['product'])55            updated_data = {56                'count_in_stock': order_item['quantity']57            }58            self._request.patch(url=update_endpoint, data=updated_data)59        logging.info(f'Updated inventory')60        # 4. Clear Cart61        delete_endpoint = self._create_endpoint(settings.CART_SERVICE, 'delete')62        delete_endpoint = delete_endpoint.format(customer=data['customer'])63        self._request.delete(url=delete_endpoint)64        logging.info(f'Cleared cart')65        logging.info('#' * 50)66        # TODO: 5. Store in redis with callback with timout of 20 min to check if67        #          order payment fails then cancel the order.68        return order69    def cancel_order(self, request, *args, **kwargs):70        data = request.data71        # 1. Update order status to Cancelled72        update_data = {73            'status': 'CL'74        }75        update_endpoint = self._create_endpoint(settings.ORDER_SERVICE, 'update')76        update_endpoint = update_endpoint.format(id=data['order'])77        order = self._request.patch(url=update_endpoint, json=update_data)78        logging.info(f'Order cancelled {order}')79        # 2. Update payment status to Refund80        update_data = {81            'status': 'RE'82        }83        update_endpoint = self._create_endpoint(settings.PAYMENT_SERVICE, 'update')84        update_endpoint = update_endpoint.format(id=order['payment'])85        payment = self._request.patch(url=update_endpoint, json=update_data)86        order['payment'] = payment87        logging.info(f'Payment updated {payment}')88        # 3. Update Inventory89        for order_item in order['order_items']:90            order_item = {91                'product': order_item['product'],92                'quantity': order_item['quantity'],93                'price': order_item['price']94            }95            update_endpoint = self._create_endpoint(settings.PRODUCT_SERVICE, 'update')96            update_endpoint = update_endpoint.format(id=order_item['product'])97            updated_data = {98                'count_in_stock': order_item['quantity']99            }100            self._request.patch(url=update_endpoint, data=updated_data)...

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.