Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _fetch_data method in autotest

Best Python code snippet using autotest_python

client.py

Source:client.py

...102            params = params if params else {}103            if "$top" not in params:104                params["$top"] = self.PAGINATION_COUNT105        return params106    def _fetch_data(self, endpoint: str, params: Optional[Dict] = None, pagination: bool = True):107        api_url = self._get_api_url(endpoint)108        params = self._get_request_params(params, pagination)109        while api_url:110            raw_response = self._make_request(api_url, params)111            value = self._get_response_value_unsafe(raw_response)112            params = None113            api_url = raw_response.get("@odata.nextLink", "")114            yield value115    def health_check(self) -> Tuple[bool, object]:116        try:117            self._get_access_token()118            return True, None119        except MsalServiceError as err:120            return False, err.args[0]121        except Exception as e:122            return False, str(e)123    def get_streams(self):124        streams = []125        for schema, method in self.ENTITY_MAP.items():126            raw_schema = json.loads(pkgutil.get_data(self.__class__.__module__.split(".")[0], f"schemas/{schema}.json"))127            streams.append(AirbyteStream(name=schema, json_schema=raw_schema, supported_sync_modes=["full_refresh"]))128        return streams129    def get_users(self):130        for users in self._fetch_data("users"):131            yield users132    def get_groups(self):133        for groups in self._fetch_data("groups"):134            yield filter(lambda item: "Team" in item["resourceProvisioningOptions"], groups)135    def _get_group_ids(self):136        if not self._group_ids:137            api_url = self._get_api_url("groups")138            params = {"$select": "id,resourceProvisioningOptions"}139            groups = self._get_response_value_unsafe(self._make_request(api_url, params=params))140            self._group_ids = [item["id"] for item in groups if "Team" in item["resourceProvisioningOptions"]]141        return self._group_ids142    def get_group_members(self):143        for group_id in self._get_group_ids():144            for members in self._fetch_data(f"groups/{group_id}/members"):145                yield members146    def get_group_owners(self):147        for group_id in self._get_group_ids():148            for owners in self._fetch_data(f"groups/{group_id}/owners"):149                yield owners150    def get_channels(self):151        for group_id in self._get_group_ids():152            for channels in self._fetch_data(f"teams/{group_id}/channels", pagination=False):153                yield channels154    def _get_channel_ids(self, group_id: str):155        api_url = self._get_api_url(f"teams/{group_id}/channels")156        # TODO: pass params={"$select": "id"} to make_request once the related bug in the MSFT API157        # is fixed: microsoftgraph/microsoft-graph-docs#11494158        channels_ids = self._get_response_value_unsafe(self._make_request(api_url))159        return channels_ids160    def get_channel_members(self):161        for group_id in self._get_group_ids():162            channels = self._get_channel_ids(group_id=group_id)163            for channel in channels:164                for members in self._fetch_data(f'teams/{group_id}/channels/{channel["id"]}/members'):165                    yield members166    def get_channel_tabs(self):167        for group_id in self._get_group_ids():168            channels = self._get_channel_ids(group_id=group_id)169            for channel in channels:170                for tabs in self._fetch_data(f'teams/{group_id}/channels/{channel["id"]}/tabs', pagination=False):171                    yield tabs172    def get_conversations(self):173        for group_id in self._get_group_ids():174            for conversations in self._fetch_data(f"groups/{group_id}/conversations"):175                yield conversations176    def _get_conversation_ids(self, group_id: str):177        api_url = self._get_api_url(f"groups/{group_id}/conversations")178        params = {"$select": "id"}179        conversation_ids = self._get_response_value_unsafe(self._make_request(api_url, params=params))180        return conversation_ids181    def get_conversation_threads(self):182        for group_id in self._get_group_ids():183            conversations = self._get_conversation_ids(group_id=group_id)184            for conversation in conversations:185                for threads in self._fetch_data(f'groups/{group_id}/conversations/{conversation["id"]}/threads'):186                    yield threads187    def _get_thread_ids(self, group_id: str, conversation_id: str):188        api_url = self._get_api_url(f"groups/{group_id}/conversations/{conversation_id}/threads")189        params = {"$select": "id"}190        thread_ids = self._get_response_value_unsafe(self._make_request(api_url, params=params))191        return thread_ids192    def get_conversation_posts(self):193        for group_id in self._get_group_ids():194            conversations = self._get_conversation_ids(group_id=group_id)195            for conversation in conversations:196                threads = self._get_thread_ids(group_id, conversation["id"])197                for thread in threads:198                    for posts in self._fetch_data(f'groups/{group_id}/conversations/{conversation["id"]}/threads/{thread["id"]}/posts'):199                        yield posts200    def get_team_drives(self):201        for group_id in self._get_group_ids():202            for drives in self._fetch_data(f"groups/{group_id}/drives"):203                yield drives204    def get_team_device_usage_report(self):205        period = self.configs["period"]206        api_url = self._get_api_url(f"reports/getTeamsDeviceUsageUserDetail(period='{period}')")207        csv_response = io.BytesIO(self._make_request(api_url))208        csv_response.readline()209        with io.TextIOWrapper(csv_response, encoding="utf-8-sig") as text_file:210            field_names = [211                "report_refresh_date",212                "user_principal_name",213                "last_activity_date",214                "is_deleted",215                "deleted_date",216                "used_web",...

domain.py

Source:domain.py

...49        return self.ingredients, self.seasonings50    # ã«ã¬ã¼ãã¼ã¿åå¾51    def _fetch_curry_ingredient(self):52        self._fetch_common_ingredient()53        self._fetch_data(self.infrastructure.get_curry_powder)54    # èããããã¼ã¿åå¾55    def _fetch_nikujyaga_ingredient(self):56        self._fetch_common_ingredient()57        self._fetch_data(self.infrastructure.get_sake)58        self._fetch_data(self.infrastructure.get_soy_sauce)59    # å±éãã¼ã¿åå¾60    def _fetch_common_ingredient(self):61        self._fetch_data(self.infrastructure.get_carrot)62        self._fetch_data(self.infrastructure.get_potato)63        self._fetch_data(self.infrastructure.get_onion)64        self._fetch_data(self.infrastructure.get_chicken)65    # ãã¼ã¿åå¾å±éå¦ç66    def _fetch_data(self, get_func: Callable[[], FoodData]):67        data = get_func()68        if data.type == FoodType.ingredient:69            self.ingredients.append(Ingredient(name=data.name, calorie=data.calorie))70        elif data.type == FoodType.seasoning:71            self.seasonings.append(Seasoning(name=data.name, calorie=data.calorie))72        else:73            raise ValueError(f"Invalid food_type of {data.type}")74@dataclass()75class Chef:76    is_water_boiled: bool = False77    # æçããã78    def cook_menu(79        self,80        ingredients: List[Ingredient],...

interactor.py

Source:interactor.py

...23            raise ValueError(f"Unsupported Menu {input_data.menu}")24        return self.ingredients, self.seasonings25    def _fetch_curry_ingredient(self):26        self._fetch_common_ingredient()27        self._fetch_data(self.repository.get_curry_powder)28        print("fetch curry ingeredients")29    def _fetch_nikujyaga_ingredient(self):30        self._fetch_common_ingredient()31        self._fetch_data(self.repository.get_sake)32        self._fetch_data(self.repository.get_soy_sauce)33        print("fetch nikujyaga ingeredients")34    def _fetch_common_ingredient(self):35        self._fetch_data(self.repository.get_carrot)36        self._fetch_data(self.repository.get_potato)37        self._fetch_data(self.repository.get_onion)38    def _fetch_data(self, get_func: Callable[[], Ingredient | Seasoning]):39        data = get_func()40        if isinstance(data, Ingredient):41            self.ingredients.append(Ingredient(name=data.name, calorie=data.calorie))42        elif isinstance(data, Seasoning):43            self.seasonings.append(Seasoning(name=data.name, calorie=data.calorie))44        else:45            raise ValueError(f"Invalid food_type of {data.food_type}")46@inject47@dataclass()48class CreateMenuInteractor(ICreateMenuUseCase):49    repository: IRepository50    presenter: ICreateMenuPresenter51    def handle(self, input_data: CreateMenuInputData):52        ingredients, seasonings = DataFetcher(self.repository).fetch(input_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.