How to use wait_for_finish method in yandex-tank

Best Python code snippet using yandex-tank

client.py

Source:client.py

1import os2import re3import json4import glob5import time6import hashlib7import logging8import urllib.parse9from . import errors10from . import requester11logger = logging.Logger('yadisk-api')12class YandexDisk(object):13    _SLEEP = 314    _requester_cls = requester.Requester15    def __init__(self, token):16        self._requester = self._requester_cls(token=token)17    def get_disk_info(self):18        """19        Get info about your disk20        Docs: https://tech.yandex.ru/disk/api/reference/capacity-docpage/21        """22        logger.info('Get disk info')23        return self._requester.get(url='disk/').json()24    def get_meta_info(25        self,26        path,27        sort=None,28        limit=None,29        offset=None,30        fields=None,31        preview_size=None,32        preview_crop=None,33        trash=False,34    ):35        """36        Get meta info about file/directory from disk or trash37        Docs: https://tech.yandex.ru/disk/api/reference/meta-docpage/38        """39        params = {40            'path': path,41            'sort': sort,42            'limit': limit,43            'offset': offset,44            'fields': fields,45            'preview_size': preview_size,46            'preview_crop': preview_crop,47        }48        logger.info('Get meta info (from trash={}) with params {}'.format(trash, params))49        return self._requester.get(50            url='disk/{}resources'.format('trash/' if trash else ''),51            params=params,52        ).json()53    def get_files_list(54        self,55        limit=None,56        offset=None,57        media_type=None,58        fields=None,59        preview_size=None,60        preview_crop=None,61    ):62        """63        Get files list from disk64        Docs: https://tech.yandex.ru/disk/api/reference/all-files-docpage/65        """66        params = {67            'limit': limit,68            'offset': offset,69            'fields': fields,70            'preview_size': preview_size,71            'preview_crop': preview_crop,72            'media_type': media_type,73        }74        logger.info('Get files list with params {}'.format(params))75        return self._requester.get(url='disk/resources/files', params=params).json()76    def get_last_uploaded(77        self,78        limit=None,79        media_type=None,80        fields=None,81        preview_size=None,82        preview_crop=None,83    ):84        """85        Get last uploaded files list86        Docs: https://tech.yandex.ru/disk/api/reference/recent-upload-docpage/87        """88        params = {89            'limit': limit,90            'fields': fields,91            'preview_size': preview_size,92            'preview_crop': preview_crop,93            'media_type': media_type,94        }95        logger.info('Get last uploaded with params {}'.format(params))96        return self._requester.get(url='disk/resources/files', params=params).json()97    def set_meta_to_resource(self, path, data, fields=None):98        """99        Set meta-data to file/directory100        Docs: https://tech.yandex.ru/disk/api/reference/meta-add-docpage/101        """102        url_params = {'path': path}103        if fields:104            url_params['fields'] = fields105        params_string = urllib.parse.urlencode(url_params, doseq=False)106        logger.info('Set meta {!r} to resource {!r}'.format(data, path))107        return self._requester.patch(108            url='disk/resources/?{}'.format(params_string),109            data=json.dumps({'custom_properties': data}),110        ).json()111    def upload_file(self, file_object, path='/', overwrite=False, skip_exists=True):112        """113        Upload file to yandex disk114        Docs: https://tech.yandex.ru/disk/api/reference/upload-docpage/115        Args:116            file_object (file): file to upload117            path (str): path to file place118            overwrite (bool): overwrite file if it exist119            skip_exists (bool): not upload file if it exists and has same hash/dates120        """121        if (122            overwrite123            and skip_exists124            and self._is_same_file(file_object, path)125        ):126            logger.debug('File {!r} already uploaded'.format(path))127            return True128        logger.info('Upload file to {!r}'.format(path))129        upload_path_url = self._requester.get(130            url='disk/resources/upload',131            params={132                'path': path,133                'overwrite': overwrite,134            },135        )136        self._requester.put(137            url=upload_path_url.json()['href'],138            files={'file': file_object},139            absolute_url=True,140        )141        return True142    def upload_directory(self, local_path, path='/', overwrite=False, skip_exists=True):143        """144        Upload all files in directory to disk145        Args:146            local_path (str): directory path on your storage147            path (str): path at yadisk148            overwrite (bool): overwrite files if it exists149            skip_exists (bool): not upload file if it exists and has same hash/dates150        """151        for is_directory, item_path, related_path in self._iter_directory_content(local_path):152            if not related_path:153                continue154            disk_path = os.path.join(path, related_path)155            if is_directory:156                try:157                    self.create_folder(path=disk_path)158                except errors.DiskPathError as exc:159                    if 'ÑÐ¶Ðµ ÑÑÑÐµÑÑÐ²ÑÐµÑ Ð¿Ð°Ð¿ÐºÐ° Ñ ÑÐ°ÐºÐ¸Ð¼ Ð¸Ð¼ÐµÐ½ÐµÐ¼' not in str(exc):160                        raise161                continue162            with open(item_path, 'rb') as f:163                self.upload_file(164                    file_object=f,165                    path=disk_path,166                    overwrite=overwrite,167                    skip_exists=skip_exists,168                )169    def upload_file_from_url(170        self,171        url,172        path,173        fields=None,174        disable_redirects=False,175        wait_for_finish=True,176        sleep=None,177    ):178        """179        Upload file from url to yandex disk180        Docs: https://tech.yandex.ru/disk/api/reference/upload-ext-docpage/181        Args:182            url (str): url to download file from it183            path (str): path to yandex disk184            fields (list[str]|None): fields in result185            disable_redirects (bool): disable redirects186            wait_for_finish (bool): wait for operation finish187            sleep (int): sleep time in seconds if need wait to finish188        """189        params_string = urllib.parse.urlencode(190            {191                'url': url,192                'path': path,193                'fields': fields,194                'disable_redirects': disable_redirects,195            },196            doseq=False,197        )198        logger.info('Upload file from url {!r} to {!r}'.format(url, path))199        return self._waiting_for_finish(200            self._requester.post(201                url='disk/resources/upload?{}'.format(params_string),202            ),203            wait_for_finish=wait_for_finish,204            sleep=sleep,205        ).json()206    def download_file(self, path, stream=False):207        """208        Download file from your disk209        Docs: https://tech.yandex.ru/disk/api/reference/content-docpage/210        Args:211            path (str): path to file212            stream (bool): stream response213        Returns:214            bytes: file content215        """216        logger.info('Download file from {!r}'.format(path))217        url_response = self._requester.get(218            url='disk/resources/download',219            params={'path': path}220        )221        return self._requester.get(222            url=url_response.json()['href'],223            absolute_url=True,224            stream=stream,225        ).content226    def copy_resource(227        self,228        from_path,229        to_path,230        overwrite=False,231        fields=None,232        wait_for_finish=True,233        sleep=None,234    ):235        """236        Copy file/directory in disk237        Docs: https://tech.yandex.ru/disk/api/reference/copy-docpage/238        Args:239            from_path (str): source path240            to_path (str): destination path241            overwrite (bool): overwrite file/directory if exists242            fields (list[str]|None): response fields list243            wait_for_finish (bool): wait for operation finish244            sleep (int): sleep time in seconds if need wait to finish245        """246        logger.info('Copy resource from {!r} to {!r}'.format(from_path, to_path))247        params_string = urllib.parse.urlencode(248            {249                'from': from_path,250                'path': to_path,251                'overwrite': overwrite,252                'fields': fields,253            },254            doseq=False,255        )256        return self._waiting_for_finish(257            self._requester.post(url='disk/resources/copy?{}'.format(params_string)),258            wait_for_finish=wait_for_finish,259            sleep=sleep,260        ).json()261    def move_resource(262        self,263        from_path,264        to_path,265        overwrite=False,266        wait_for_finish=True,267        sleep=None,268    ):269        """270        Move file/directory in disk271        Docs: https://tech.yandex.ru/disk/api/reference/move-docpage/272        Args:273            from_path (str): source path274            to_path (str): destination path275            overwrite (bool): overwrite file/directory if exists276            wait_for_finish (bool): wait for operation finish277            sleep (int): sleep time in seconds if need wait to finish278        """279        logger.info('Move resource from {!r} to {!r}'.format(from_path, to_path))280        params_string = urllib.parse.urlencode(281            {282                'from': from_path,283                'path': to_path,284                'overwrite': overwrite,285            },286            doseq=False,287        )288        return self._waiting_for_finish(289            self._requester.post(url='disk/resources/move?{}'.format(params_string)),290            wait_for_finish=wait_for_finish,291            sleep=sleep,292        ).json()293    def delete_resource(self, path, permanently=False, wait_for_finish=True, sleep=None):294        """295        Remove file/directory to trash or at all296        Docs: https://tech.yandex.ru/disk/api/reference/delete-docpage/297        Args:298            path (str): path to file/directory299            permanently (bool): true if your want remove resource without trash300            wait_for_finish (bool): wait for operation finish301            sleep (int): sleep time in seconds if need wait to finish302        """303        logger.info('Delete resource from {!r}'.format(path))304        params_string = urllib.parse.urlencode(305            {306                'path': path,307                'permanently': permanently,308            },309            doseq=False,310        )311        self._waiting_for_finish(312            self._requester.delete(313                url='disk/resources?{}'.format(params_string)314            ),315            wait_for_finish=wait_for_finish,316            sleep=sleep,317        )318        return True319    def create_folder(self, path, fields=None):320        """321        Create folder in your disk322        Docs: https://tech.yandex.ru/disk/api/reference/create-folder-docpage/323        """324        logger.info('Create folder {!r}'.format(path))325        params_string = urllib.parse.urlencode(326            {327                'path': path,328                'fields': fields,329            },330            doseq=False,331        )332        return self._requester.put(url='disk/resources/?{}'.format(params_string)).json()333    def publish_resource(self, path):334        """335        Publish_resource336        Docs: https://tech.yandex.ru/disk/api/reference/publish-docpage/337        Args:338            path (str): path to resource339        """340        logger.info('Publish resource {!r}'.format(path))341        params_string = urllib.parse.urlencode({'path': path})342        return self._requester.put(343            url='disk/resources/publish?{}'.format(params_string)344        ).json()345    def unpublish_resource(self, path):346        """347        Unpublish resource348        Docs: https://tech.yandex.ru/disk/api/reference/publish-docpage/#unpublish-q349        Args:350            path (str): path to resource351        """352        logger.info('Unpublish resource {!r}'.format(path))353        params_string = urllib.parse.urlencode({'path': path})354        return self._requester.put(355            url='disk/resources/unpublish?{}'.format(params_string)356        ).json()357    def empty_trash(self, path=None, wait_for_finish=True, sleep=None):358        """359        Empty trash or delete resource from trash360        Docs: https://tech.yandex.ru/disk/api/reference/trash-delete-docpage/361        """362        logger.info('Empty trash')363        path_param = '?{}'.format(364            urllib.parse.urlencode({'path': path})365        ) if path else ''366        self._waiting_for_finish(367            self._requester.delete(url='disk/trash/resources/{}'.format(path_param)),368            wait_for_finish=wait_for_finish,369            sleep=sleep,370        )371        return True372    def restore_from_trash(self, path, name=None, overwrite=False):373        """374        Restore resource from trash375        Docs: https://tech.yandex.ru/disk/api/reference/trash-restore-docpage/376        """377        logger.info('Restore {!r} from trash'.format(path))378        params_string = urllib.parse.urlencode(379            {380                'path': path,381                'name': name,382                'overwrite': overwrite,383            },384            doseq=False,385        )386        return self._waiting_for_finish(387            self._requester.put(388                url='disk/trash/resources/restore?{}'.format(params_string),389            )390        ).json()391    def _waiting_for_finish(self, response, wait_for_finish=True, sleep=None):392        """393        Waiting for finish operation, if you want394        Args:395            response (requests.Response): response object396            wait_for_finish (bool): wait for operation finish397            sleep (int): sleep time in seconds if need wait to finish398        Returns:399            requests.Response400        """401        if (402            wait_for_finish403            and response.status_code == requester.STATUS_ACCEPTED404        ):405            check_status_url = response.json()['href']406            sleep = sleep or self._SLEEP407            while True:408                time.sleep(sleep)409                response = self._requester.get(check_status_url, absolute_url=True)410                if (411                    response.status_code == requester.STATUS_OK412                    and response.json()['status'] == 'success'413                ):414                    return response415        return response416    def _iter_directory_content(self, path, start_path=None):417        """418        All files generator419        Args:420            path (str): path to find all files421        Yields:422            bool, str, str:423                1. is directory flag424                2. local path to file/dir425                3. related path to file/dir426        """427        if start_path is None:428            start_path = path429        path_list = glob.glob(path)430        for path_item in path_list:431            related_path = re.sub(r'{}\/?\*?'.format(start_path), '', path_item)432            is_directory = os.path.isdir(path_item)433            yield is_directory, path_item, related_path434            yield from self._iter_directory_content(os.path.join(path_item, '*'), start_path=start_path)435    def _is_same_file(self, fileobject, path):436        """437        Exists same file438        """439        # check file changed440        try:441            file_meta_info = self.get_meta_info(path=path, fields=['md5'])442        except errors.NotFoundError:443            return False444        current_file_md5 = hashlib.md5(fileobject.read()).hexdigest()...

socrata.py

Source:socrata.py

...103        output_schema = add_geometry_to_output_schema(output_schema, geometry)104    # Handle pre-1.x versions of Socrata-py105    if isinstance(output_schema, tuple):106        _, output_schema = output_schema107    output_schema.wait_for_finish()108    job: Job = revision.apply(output_schema=output_schema)109    if wait_for_finish is True:110        job.wait_for_finish()111    return revision112def update_existing_dataset(113    client: Socrata, dataframe: DataFrame, dataset_id: str, *, wait_for_finish: bool = False114):115    """Use a dataframe to update an existing Socrata dataset."""116    view = client.views.lookup(dataset_id)117    # Handle pre-1.x versions of Socrata-py118    if isinstance(view, tuple):119        ok, view = view120        ok, revision = view.revisions.create_replace_revision()121        ok, upload = revision.create_upload('autocensus-update')122        ok, source = upload.df(dataframe)123        source.wait_for_finish()124        output = source.get_latest_input_schema().get_latest_output_schema()125        output.wait_for_finish()126        job = revision.apply(output_schema=output)127    else:128        revision = view.revisions.create_replace_revision()129        upload = revision.create_upload('autocensus-update')130        source = upload.df(dataframe)131        source.wait_for_finish()132        output = source.get_latest_input_schema().get_latest_output_schema()133        output.wait_for_finish()134        job = revision.apply(output_schema=output)135    if wait_for_finish is True:136        job.wait_for_finish()137    return revision138def build_dataset_name(estimate: int, years: Iterable) -> str:139    """Produce a nicely formatted dataset name."""140    unique_years = sorted(set(years))141    if len(unique_years) > 1:142        years_range = f'{min(unique_years)}â{max(unique_years)}'143    else:144        years_range = unique_years[0]145    query_name = f'American Community Survey {estimate}-Year Estimates, {years_range}'146    return query_name147def to_socrata(148    domain: Union[URL, str],149    dataframe: DataFrame,150    dataset_id: Optional[str] = None,...

pipette_foreach.py

Source:pipette_foreach.py

...23            msg = self.serial_connection.serial_port.readline()24            if msg:25                print("Got response.", msg)26                return msg27    def wait_for_finish(self):28        query_cmd = "/1Q\r"29        i = 130        while True:31            self.serial_connection.send_commond_string_foreach(query_cmd)32            time.sleep(WAIT_TIME)33            res = self.serial_connection.serial_port.readline()34            if not (b'/0@\x03' in res):35                print(i, res)36            i = i+137            if i > 300:38                print("Pipette no response for 30 seconds, system exited")39                # sys.exit()40            if b'/0`' in res:41                print("Finish request.")42                return "finish"43    def send_commond_string_foreach(self, cmd):44        self.wait_for_finish()45        self.serial_connection.send_commond_string_foreach(cmd)46    def initialization(self):47        init_cmd = '/1ZR\r'48        self.send_commond_string_foreach(init_cmd)49        time.sleep(1)50        self.wait_for_response()51        self.wait_for_finish()52        print("initialization finished.")53        normal_mode = "/1N0R\r"54        # N0 mode and N1 mode55        self.send_commond_string_foreach(normal_mode)56        self.wait_for_response()57        self.wait_for_finish()58    def increase_range(self):59        cmd = '/1u1,3500R\r'60        self.send_commond_string_foreach(cmd)61        self.wait_for_response()62        self.wait_for_finish()63    def aspirate(self, volume):64        '''aspirate volume = ? uL'''65        mL_per_step = 0.31966        steps = int(volume/mL_per_step)67        aspirate_cmd = "/1P" + str(steps) + "R\r"68        self.send_commond_string_foreach(aspirate_cmd)69        self.wait_for_response()70        self.wait_for_finish()71    def send_drop_tip_cmd(self):72        eject_tip = "/1ER\r"73        self.send_commond_string_foreach(eject_tip)74        time.sleep(0.5)75        self.wait_for_response()76        self.wait_for_finish()77    def dispense(self, volume=0):  # volume in uL78        dispense_all = "/1A0R\r"79        self.send_commond_string_foreach(dispense_all)80        self.wait_for_response()81        self.wait_for_finish()82    def set_speed(self, speed):83        '''speed = 0-40, default = 11, bigger nmuber is slower '''84        cmd = "/1S" + str(speed) + "R\r"85        self.send_commond_string_foreach(cmd)86        self.wait_for_response()87        self.wait_for_finish()88    def is_tip_attached(self):89        check_tip_cmd = "/1?31R\r"90        self.send_commond_string_foreach(check_tip_cmd)91        res = self.wait_for_response()92        self.wait_for_finish()93        if b"`1\x03" in res:94            print("tip attached! code:", res)95            return True96        else:97            print("tip NOT attached! code:", res)98            return False99    def send_pickup_tip_cmd(self):  # do nothing, not needed for foreach model100        pass101    def set_transport_air_volume(self, volume=20):  # volume in uL102        if volume > 0:...

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.