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

How to use invoke_rest_api method in localstack

Best Python code snippet using localstack_python

pubg_api.py

Source:pubg_api.py

...26        # holders for response variables, for rate-limit avoidance purposes27        self.response_status_code = None28        self.response_headers = None29        return None30    def invoke_rest_api(self, url, headers, params=None):31        """32        All the calls to the API are done through this function, so that any necessary changes can easily33        be made in one place34        """35        # rate limiting goes here. Check the latest response header dict and36        # see if it has the rate limit headers. If it does and we're out of 37        # calls, wait till the reset. Otherwise, just run the func.38        # if we've actually run something already39        if self.response_status_code is not None:40            # and it has a rate limit41            if 'X-Ratelimit-Remaining' in self.response_headers.keys():42                # if the rate limit is 043                if self.response_headers['X-Ratelimit-Remaining'] == '0':44                    # rather than a bare sleep we'll while away the time till current > reset time, so that in the case45                    # we end up waiting past the reset time of the last call due to db writes or whatever we're not unecessarily 46                    # hanging around.47                    reset_time = float(self.response_headers['X-Ratelimit-Reset'])48                    # log here since this is the first place we're _actually_ waiting49                    logging.debug('Waiting {0}s to avoid rate limiter'.format((reset_time-time.time()) + 5.))50                    while time.time() < (reset_time + 5):51                        continue52        # our waiting is done, on with the call, yo!53        r = requests.get(54            url=url,55            headers=headers,56            params=params57        )58        # store the latest response code and headers59        self.response_status_code = r.status_code60        self.response_headers = r.headers61        return r62    def get_players(self):63        module = '/players'64        i = 065        while i < len(self.player_names):66            payload = {'filter[playerNames]': ','.join(self.player_names[i:i+10])67            }68            logging.debug("get_players:Payload = [{0}]".format(','.join(self.player_names[i:i+10])))69            try:70                r = self.invoke_rest_api(71                    url=self.base_url + self.shard + module,72                    headers=self.headers,73                    params=payload74                    )75            except Exception as e:76                logging.exception("get_players: API Request")77            try:78                self.players = self.players + r.json()['data']79            except Exception as e:80                logging.exception("get_players: Append to players")81            i += 1082        return True83    def get_matches(self, process_matches):84        with multiprocessing.Pool() as pool:85            multiprocessed_matches = pool.map(self.get_match, process_matches)86        self.matches = [i for j in multiprocessed_matches for i in j]87        logging.info("get_matches: Num Matches fetched = {0}".format(len(self.matches)))88        return True89    def get_match(self, match_id):90        """91        Fetch a single match by calling the PUBG API, and append it to the list92        of matches93        """94        matches = []95        logging.debug("get_match: match_id={0}".format(match_id))96        module = '/matches/{0}'.format(match_id)97        try:98            r = self.invoke_rest_api(99                url=self.base_url + self.shard + module,100                headers=self.headers101            )102        except Exception as e:103            logging.exception("get_match: API Request:")104        try:105            matches.append(r.json())106        except:107            logging.exception("get_match: Append to matches")108        return matches109    def get_seasons(self):110        """111        Fetch the list of seasons from the API. This shouldn't change more than112        once a month and the API documentation specifically says not to call the113        endpoint more regularly than that, so we'll cache it in the DB unless114        the time-last-called is more than 1 month ago.115        """116        module = '/seasons'117        r = self.invoke_rest_api(118            url=self.base_url + self.shard + module,119            headers=self.headers120        )121        try:122            self.seasons = r.json()['data']123        except:124            logging.exception('get_seasons: Append data to seasons')125        return None126    def get_current_season(self):127        """128        Returns the current season only from the cached list of seasons129        """130        return [season for season in self.seasons if season['attributes']['isCurrentSeason']]131    def get_player_ranked_season_stats(self, combo):132        """133        Fetches the ranked player season stats. Combo is a tuple consisting of134        (player_id, season_id).135        """136        module = '/players/{0}/seasons/{1}/ranked'.format(137            combo[0],138            combo[1]139        )140        r = self.invoke_rest_api(141            url=self.base_url + self.shard.split('-')[0] + module,142            headers=self.headers143        )144        logging.debug("get_player_ranked_season_stats: {0}: {1}: {2}".format(combo[0], combo[1], json.dumps(r.json(), indent=4)))145        if r.status_code == 200:146            try:147                self.player_ranked_season_stats.append(148                    r.json()['data']149                )150            except:151                logging.exception("get_player_ranked_season_stats: Error appending data to list")152        else:153            logging.debug("get_player_ranked_season_stats returned something other than HTTP 200")154        return None155    def get_player_season_stats(self, combo):156        """157        This call has to be rate limited, as it's pretty fast to complete which158        means it often hits the API's transaction limit and returns a html 429159        code. 4 second pauses between calls seems to avoid the issue, but I'll160        include a condition to pause for 20 seconds if it hits the issue.161        combo is a (player_id, season_id) tuple162        """163        module ='/players/{0}/seasons/{1}'.format(164            combo[0],165            combo[1]166        )167        r = self.invoke_rest_api(168            url=self.base_url + self.shard + module,169            headers=self.headers170        )171        while r.status_code == 429:172            time.sleep(10)173            r = self.invoke_rest_api(174                url=self.base_url + self.shard + module,175                headers=self.headers176            )177        logging.debug("get_player_season_stats: {0}: {1}: {2}".format(combo[0], combo[1], json.dumps(r.json(), indent=4)))178        if r.status_code == 200:179            try:180                self.player_season_stats.append(181                    r.json()['data']182                )183            except:184                logging.exception("get_player_season_stats: Error appending data to list")185        else:186            logging.debug("get_player_season_stats returned something other than HTTP 200")187        return None188    def get_player_lifetime_stats(self, process_players):189        for player in process_players:190            module = '/players/{0}/seasons/lifetime'.format(191                player192            )193            r = self.invoke_rest_api(194                url=self.base_url + self.shard + module,195                headers=self.headers196            )197            while r.status_code == 429:198                time.sleep(10)199                r = self.invoke_rest_api(200                    url=self.base_url + self.shard + module,201                    headers=self.headers202                )203            if r.status_code == 200:204                try:205                    self.player_lifetime_stats.append(206                        r.json()['data']207                    )208                except:209                    logging.exception("get_player_lifetime_stats: Error appending data to list")210            else:211                logging.debug("get_player_lifetime_stats returned something other than HTTP 200")...

router_asf.py

Source:router_asf.py

...103        self.router.add(104            "/restapis/<api_id>/<stage>/_user_request_/<path:path>",105            endpoint=self.invoke_rest_api,106        )107    def invoke_rest_api(self, request: Request, **url_params: Dict[str, Any]) -> Response:108        if not get_api_region(url_params["api_id"]):109            return Response(status=404)110        invocation_context = to_invocation_context(request, url_params)111        result = invoke_rest_api_from_request(invocation_context)112        if result is not None:113            return convert_response(result)...

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.