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

How to use _fetch_records method in localstack

Best Python code snippet using localstack_python

application.py

Source:application.py

...49def user_ratings(user_id):50    """51    List all of the ratings of a single user.52    """53    return _fetch_records(f"SELECT item_id, rating_type FROM ratings WHERE user_id = {user_id}")54@application.route('/users/<user_id>/counts')55def user_counts(user_id):56    """57    Count number of ratings per type, for a single user.58    Does not include a rating type in the list if its count is zero.59    """60    return _fetch_records(f"SELECT rating_type, count FROM counts_by_rating_type WHERE user_id = {user_id} AND count > 0")61def _fetch_records(query):62    """63    Helper private method for querying the DB64    """65    con = connect()66    cursor = con.cursor()67    cursor.execute(query)68    row_headers = [x[0] for x in cursor.description]  # this will extract row headers69    results = cursor.fetchall()70    json_data = []71    for result in results:72        json_data.append(dict(zip(row_headers, result)))73    cursor.close()74    return json.dumps(json_data)75@application.route('/users/<user_id>', methods=['DELETE'])...

client.py

Source:client.py

...16class DOABOAIClient():17    def __init__(self):18        self._sickle = Sickle(const.DOAB_OAI_ENDPOINT)19    def fetch_records_for_publisher_id(self, publisher_id):20        return self._fetch_records(publisher_id=publisher_id)21    def fetch_all_records(self):22        return self._fetch_records()23    def _fetch_records(self, publisher_id=None):24        kwargs = {25            "metadataPrefix": "oai_dc",26        }27        if publisher_id is not None:28            kwargs["set"] = f"publisher_{publisher_id}"29        return (30            DOABRecord(record)31            for record in self._sickle.ListRecords(**kwargs)32            if record_is_active_book(record)33        )34class DOABRecord():35    """ A wrapper for an OAI record extracted from DOAB """36    def __init__(self, record):37        self.metadata = record.metadata...

pandas-wrapping.py

Source:pandas-wrapping.py

...6        self.apiKey = apiKey7        self.apiVersion = apiVersion8        self.baseId = baseId9        self.endpoint = endpoint10    def _fetch_records(self, url, headers):11        response = requests.get(url, headers=headers)12        if response.status_code != 200:13            raise ValueError(f"Error: {response.status_code} - {response.reason}")14        payload = response.json()15        offset = None16        if 'offset' in payload.keys():17            offset = f"?offset={payload['offset']}"18        return response, offset19    def _transform_dataframe4sink(self, data):20        fields = data.fillna('').to_dict('records')21        fields = [{k: v for k, v in d.items() if v and v.strip()} for d in fields]22        dataframe = pd.DataFrame(data={'fields': fields})23        records = dict({'records': dataframe.to_dict(orient='records')})24        sink = json.dumps(records)25        return sink26    def query_table(self, endpoint):27        url = f"https://api.airtable.com/{self.apiVersion}/{self.baseId}/{endpoint}"28        headers = {29            'Authorization': f'Bearer {self.apiKey}',30            'Content-Type': 'application/json'31        }32        records = list()33        offset = ""34        while (True):35            response, offset = self._fetch_records(url + offset, headers=headers)36            records += response.json()['records']37            if not (offset):38                break39        dataframe = pd.json_normalize(records)40        return dataframe41    def insert_records(self, endpoint, dataframe, pagination_max=10):42        url = f"https://api.airtable.com/{self.apiVersion}/{self.baseId}/{endpoint}"43        headers = {44            'Authorization': f'Bearer {self.apiKey}',45            'Content-Type': 'application/json',46        }47        log = list()48        for i in range(0, dataframe.shape[0], pagination_max):49            payload = dataframe[i:(i + pagination_max)].copy()...

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.