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

How to use get_next_error_id method in refurb

Best Python code snippet using refurb_python

es_util.py

Source:es_util.py

...182                    cur_low_counts += 1183                time.sleep(1)184                sys.stderr.flush()185    @synchronized186    def get_next_error_id(self):187        self.error_id_counter += 1188        return self.error_id_counter189    @synchronized190    def update_pb(self, start, success=False):191        if start:192            self.submission_count += 1193            self.submission_pb.start(max_value=self.submission_count)194            self.submission_pb.update(self.submission_completed_count)195        else:196            if success:197                self.submission_completed_count += 1198            self.submission_pb.update(self.submission_completed_count)199    # noinspection PyBroadException200    def submit_bulk(self, idx_name, actions):201        self.update_pb(True)202        if self.complete_futures:203            complete_futures_values(actions)204        error_id = None205        success = False206        for try_i in range(4):207            if self.stop_submission:208                return209            try:210                helpers.bulk(client=self.es_util_owner.es_conn, actions=actions)211                success = True212                break213            except:214                if error_id is None:215                    error_id = self.get_next_error_id()216                print("ERROR ON SUBMISSION TO ELASTIC!! TRY: {0} ERROR_ID: {1} AT: {2}"217                      .format(try_i, error_id, time.strftime("%a, %d %b %Y %H:%M:%S +0000")),218                      file=sys.stderr)219                with open('./ERROR_{0}_try_{1}.trace'.format(error_id, try_i), 'w') as error_file:220                    error_file.write(traceback.format_exc()+'\n')221                with open('./ERROR_{0}_try_{1}_request.py'.format(error_id, try_i), 'w') as error_req_file:222                    error_req_file.write('request = \\\n')223                    error_req_file.write(pprint.pformat(actions))224                sys.stderr.flush()225                # Retries after giving the server a small break226                time.sleep(3**(try_i+1))227        self.update_counts(idx_name, success, len(actions))228        self.update_pb(False, success)229# ----------------------------------------------------------------------------------------------------------------------...

gen.py

Source:gen.py

...53    if path.is_relative_to(cwd):54        to_remove = len(cwd.parents) + 155        return [path, *list(path.parents)[:-to_remove]]56    return []57def get_next_error_id(prefix: str) -> int:58    highest = 059    for module in get_modules([]):60        if error := get_error_class(module):61            error_code = ErrorCode.from_error(error)62            if error_code.prefix == prefix:63                highest = max(highest, error_code.id + 1)64    return highest65NODES: dict[str, type] = {x.__name__: x for x in METHOD_NODE_MAPPINGS.values()}66def node_type_prompt() -> list[str]:67    return sorted(68        fzf(69            list(NODES.keys()), args=["--prompt", "type> ", "--multi"]70        ).splitlines()71    )72def filename_prompt() -> Path:73    return Path(74        fzf(75            None,76            args=[77                "--prompt",78                "filename> ",79                "--print-query",80                "--query",81                "refurb/checks/",82            ],83        ).splitlines()[0]84    )85def prefix_prompt() -> str:86    return fzf(87        [""], args=["--prompt", "prefix> ", "--print-query", "--query", "FURB"]88    )89def build_imports(names: list[str]) -> str:90    modules: defaultdict[str, list[str]] = defaultdict(list)91    for name in names:92        modules[NODES[name].__module__].append(name)93    return "\n".join(94        f"from {module} import {', '.join(names)}"95        for module, names in sorted(modules.items(), key=lambda x: x[0])96    )97def main() -> None:98    selected = node_type_prompt()99    file = filename_prompt()100    if file.suffix != ".py":101        print('refurb: File must end in ".py"')102        sys.exit(1)103    prefix = prefix_prompt()104    template = FILE_TEMPLATE.format(105        accept_type=" | ".join(selected),106        imports=build_imports(selected),107        prefix=prefix,108        id=get_next_error_id(prefix) or 100,109        pattern=" | ".join(f"{x}()" for x in selected),110    )111    with suppress(FileExistsError):112        file.parent.mkdir(parents=True, exist_ok=True)113        for folder in folders_needing_init_file(file.parent):114            (folder / "__init__.py").touch(exist_ok=True)115    file.write_text(template, "utf8")116    print(f"Generated {file}")117if __name__ == "__main__":...

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.