Function is_artifact_locked

ci/util/cache_lock.py:205–231

Check if an artifact is currently locked. Args: cache_dir: Global cache directory url: URL to check Returns: True if artifact is locked by an active process, False otherwise

(cache_dir: Path, url: str)

def is_artifact_locked(cache_dir: Path, url: str) -> bool:
    """
    Check if an artifact is currently locked.

    Args:
        cache_dir: Global cache directory
        url: URL to check

    Returns:
        True if artifact is locked by an active process, False otherwise
    """
    cache_key = str(sanitize_url_for_path(url))
    artifact_dir = cache_dir / cache_key
    lock_file = artifact_dir / ".download.lock"

    db = get_lock_database(lock_file)
    lock_name = str(lock_file)

    # Check if held in database
    if not db.is_held(lock_name):
        return False

    # Check if it's stale
    if db.is_lock_stale(lock_name):
        return False

    return True