feat: allow user to pass in tuple of exceptions to skipbadfiles

[ikrommyd]

cc @alexander-held

[ikrommyd]

I'm interested in what y'all think of the algorithm which currently reads:

    def automatic_retries(
        retries: int,
        skipbadfiles: Union[bool, tuple[type[BaseException], ...]],
        func,
        *args,
        **kwargs,
    ):
        import warnings

        if not isinstance(skipbadfiles, tuple) and skipbadfiles is True:
            skipbadfiles = (OSError,)

        retry_count = 0
        while retry_count <= retries:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                chain = _exception_chain(e)
                if (
                    skipbadfiles
                    and (retries == retry_count)
                    and any(isinstance(c, skipbadfiles) for c in chain)
                ):
                    warnings.warn(str(e))
                    break
                else:
                    raise e
                warnings.warn("Attempt %d of %d." % (retry_count + 1, retries + 1))
            retry_count += 1

retries is being set to zero automatically for dask executor because dask already has automatric retries built in and defaults to 3 for other executors.

[ikrommyd]

@alexander-held
Just to finish the thread above.

from coffea import processor


def analysis(events):
    raise ValueError("foo")


if __name__ == "__main__":
    fileset = {
        "sample": [
            "tests/samples/nano_dy.root"
        ]
    }

    executor = processor.IterativeExecutor(compression=None)
    run = processor.Runner(executor=executor, chunksize=20, skipbadfiles=(Exception,))
    out = run(processor_instance=analysis, fileset=fileset, treename="Events")
    print(out)

This works just fine. It catches the error. If you say skipbadfiles=True, it does not because it's not an OSError (default).