Repository: psf/black
File: src/black_primer/lib.py (Python)

Code before:
    except (asyncio.TimeoutError, CalledProcessError) as e:
        LOG.error(f"Unable to git clone / pull {project_config['git_clone_url']}: {e}")
        return None

    return repo_path


def handle_PermissionError(
    func: Callable, path: Path, exc: Tuple[Any, Any, Any]
) -> None:
    """
    Handle PermissionError during shutil.rmtree.

    This checks if the erroring function is either 'os.rmdir' or 'os.unlink', and that
    the error was EACCES (i.e. Permission denied). If true, the path is set writable,
    readable, and executable by everyone. Finally, it tries the error causing delete
    operation again.

    If the check is false, then the original error will be reraised as this function
    can't handle it.
    """
    excvalue = exc[1]
    LOG.debug(f"Handling {excvalue} from {func.__name__} ... ")
    if func in (os.rmdir, os.unlink) and excvalue.errno == errno.EACCES:
        LOG.debug(f"Setting {path} writable, readable, and executable by everyone... ")
        os.chmod(path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)  # chmod 0777
        func(path)  # Try the error causing delete operation again
    else:
        raise


async def load_projects_queue(
    config_path: Path,
) -> Tuple[Dict[str, Any], asyncio.Queue]:
    """Load project config and fill queue with all the project names"""
    with config_path.open("r") as cfp:
        config = json.load(cfp)

    # TODO: Offer more options here
    # e.g. Run on X random packages or specific sub list etc.
    project_names = sorted(config["projects"].keys())
    queue: asyncio.Queue = asyncio.Queue(maxsize=len(project_names))
    for project in project_names:
        await queue.put(project)

    return config, queue


async def project_runner(
    idx: int,
    config: Dict[str, Any],

Review comment (suggestion):
If you want to remove the `stat` dependency, you could just do the below, although I think the `stat` constants are more descriptive :)
```suggestion
        os.chmod(path, 511)  # chmod 0777
```


Code after:
    except (asyncio.TimeoutError, CalledProcessError) as e:
        LOG.error(f"Unable to git clone / pull {project_config['git_clone_url']}: {e}")
        return None

    return repo_path


def handle_PermissionError(
    func: Callable, path: Path, exc: Tuple[Any, Any, Any]
) -> None:
    """
    Handle PermissionError during shutil.rmtree.

    This checks if the erroring function is either 'os.rmdir' or 'os.unlink', and that
    the error was EACCES (i.e. Permission denied). If true, the path is set writable,
    readable, and executable by everyone. Finally, it tries the error causing delete
    operation again.

    If the check is false, then the original error will be reraised as this function
    can't handle it.
    """
    excvalue = exc[1]
    LOG.debug(f"Handling {excvalue} from {func.__name__}... ")
    if func in (os.rmdir, os.unlink) and excvalue.errno == errno.EACCES:
        LOG.debug(f"Setting {path} writable, readable, and executable by everyone... ")
        os.chmod(path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)  # chmod 0777
        func(path)  # Try the error causing delete operation again
    else:
        raise


async def load_projects_queue(
    config_path: Path,
) -> Tuple[Dict[str, Any], asyncio.Queue]:
    """Load project config and fill queue with all the project names"""
    with config_path.open("r") as cfp:
        config = json.load(cfp)

    # TODO: Offer more options here
    # e.g. Run on X random packages or specific sub list etc.
    project_names = sorted(config["projects"].keys())
    queue: asyncio.Queue = asyncio.Queue(maxsize=len(project_names))
    for project in project_names:
        await queue.put(project)

    return config, queue


async def project_runner(
    idx: int,
    config: Dict[str, Any],