diff --git a/CHANGELOG.md b/CHANGELOG.md index 3fc578ec6..927e53063 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## [1.7.4] + +### Enhancements + +- **feat(PLU-441): ready the GitHub source connector for Foundation.** Adds OAuth support (`oauth_token`/`refresh_token` alongside the existing PAT, refresh handled upstream per PLU-381) and emits the git blob SHA as `metadata.version` for incremental change detection. Indexing fetches the repo once per run instead of twice per file, falls back to a per-directory walk when GitHub truncates the recursive tree, and skips empty files, directories, and submodules. Fixes downloads failing with `FileNotFoundError` on nested paths. GitHub API errors are classified consistently (auth → `UserAuthError`, throttle → `RateLimitError`, 5xx → `ProviderError`), and transient throttles / 5xx / network errors are retried with exponential backoff that honors GitHub's `Retry-After` (`max_retries`, default 10). + ## [1.7.3] ### Fixes diff --git a/pyproject.toml b/pyproject.toml index dfa07b24e..ffdf906be 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,7 +52,7 @@ dropbox = ["dropboxdrivefs", "fsspec"] duckdb = ["pandas", "duckdb"] elasticsearch = ["elasticsearch[async]<9.0.0"] gcs = ["gcsfs", "fsspec", "beautifulsoup4"] -github = ["pygithub>1.58.0", "requests"] +github = ["pygithub>1.58.0", "requests", "tenacity"] gitlab = ["python-gitlab"] google-drive = ["google-api-python-client", "tenacity"] hubspot = ["hubspot-api-client", "urllib3"] diff --git a/test/integration/connectors/test_github.py b/test/integration/connectors/test_github.py index 7be8531d1..4daa482de 100644 --- a/test/integration/connectors/test_github.py +++ b/test/integration/connectors/test_github.py @@ -20,16 +20,7 @@ ) -@pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, UNCATEGORIZED_TAG) -@pytest.mark.asyncio -@requires_env("GH_READ_ONLY_ACCESS_TOKEN") -async def test_github_source(temp_dir): - access_token = os.environ["GH_READ_ONLY_ACCESS_TOKEN"] - connection_config = GithubConnectionConfig( - access_config=GithubAccessConfig(access_token=access_token), - url="dcneiner/Downloadify", - ) - +async def _validate_github_source(connection_config: GithubConnectionConfig, temp_dir): indexer = GithubIndexer( connection_config=connection_config, index_config=GithubIndexerConfig(file_glob=["*.txt", "*.html"]), @@ -52,3 +43,29 @@ async def test_github_source(temp_dir): postdownload_file_data_check=source_filedata_display_name_set_check, ), ) + + +@pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, UNCATEGORIZED_TAG) +@pytest.mark.asyncio +@requires_env("GH_READ_ONLY_ACCESS_TOKEN") +async def test_github_source(temp_dir): + access_token = os.environ["GH_READ_ONLY_ACCESS_TOKEN"] + connection_config = GithubConnectionConfig( + access_config=GithubAccessConfig(access_token=access_token), + url="dcneiner/Downloadify", + ) + await _validate_github_source(connection_config, temp_dir) + + +@pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, UNCATEGORIZED_TAG) +@pytest.mark.asyncio +@requires_env("GH_OAUTH_ACCESS_TOKEN") +async def test_github_source_oauth(temp_dir): + # The platform's init-oauth-refresh container populates oauth_token with a fresh access + # token before each run; here we supply an already-valid OAuth token to exercise the path. + oauth_token = os.environ["GH_OAUTH_ACCESS_TOKEN"] + connection_config = GithubConnectionConfig( + access_config=GithubAccessConfig(oauth_token=oauth_token), + url="dcneiner/Downloadify", + ) + await _validate_github_source(connection_config, temp_dir) diff --git a/test/unit/connectors/test_github.py b/test/unit/connectors/test_github.py new file mode 100644 index 000000000..5b748b6fc --- /dev/null +++ b/test/unit/connectors/test_github.py @@ -0,0 +1,826 @@ +import sys +import types +from time import time +from types import SimpleNamespace +from unittest import mock + +import pytest + +from unstructured_ingest.data_types.file_data import FileData +from unstructured_ingest.error import ( + ProviderError, + RateLimitError, + SourceConnectionNetworkError, + UnstructuredIngestError, + UserAuthError, + UserError, + ValueError, +) +from unstructured_ingest.processes.connectors.github import ( + _MAX_RETRY_AFTER_WAIT, + GithubAccessConfig, + GithubConnectionConfig, + GithubDownloader, + GithubDownloaderConfig, + GithubIndexer, + GithubIndexerConfig, + _GitFile, + _honor_retry_after, + _parse_retry_after, + _rate_limit_backoff, +) + +REPO = "dcneiner/Downloadify" + + +def _connection_config() -> GithubConnectionConfig: + return GithubConnectionConfig( + access_config=GithubAccessConfig(access_token="pat-token"), + url=REPO, + ) + + +def test_connection_config_pat_auth(): + """Existing PAT (access_token) configuration continues to work unchanged.""" + config = GithubConnectionConfig( + access_config=GithubAccessConfig(access_token="pat-token"), + url=REPO, + ) + assert config.access_config.get_secret_value().access_token == "pat-token" + + +def test_connection_config_oauth_auth(): + """OAuth configuration (oauth_token + refresh_token) is accepted.""" + config = GithubConnectionConfig( + access_config=GithubAccessConfig( + oauth_token="oauth-token", + refresh_token="refresh-token", + ), + url=REPO, + ) + access_configs = config.access_config.get_secret_value() + assert access_configs.oauth_token == "oauth-token" + assert access_configs.refresh_token == "refresh-token" + + +def test_connection_config_no_auth(): + with pytest.raises(ValueError): + GithubConnectionConfig(access_config=GithubAccessConfig(), url=REPO) + + +def test_connection_config_pat_and_oauth_are_exclusive(): + with pytest.raises(ValueError): + GithubConnectionConfig( + access_config=GithubAccessConfig( + access_token="pat-token", + oauth_token="oauth-token", + ), + url=REPO, + ) + + +def _client_token(access_config: GithubAccessConfig) -> str: + """Build the client with the github module mocked and return the token passed to it.""" + github_module = mock.MagicMock() + with mock.patch.dict(sys.modules, {"github": github_module}): + config = GithubConnectionConfig(access_config=access_config, url=REPO) + config.get_client() + _, kwargs = github_module.Github.call_args + return kwargs["login_or_token"] + + +def test_get_client_uses_access_token_for_pat(): + assert _client_token(GithubAccessConfig(access_token="pat-token")) == "pat-token" + + +def test_get_client_uses_oauth_token_when_present(): + assert ( + _client_token(GithubAccessConfig(oauth_token="oauth-token", refresh_token="refresh-token")) + == "oauth-token" + ) + + +# --------------------------------------------------------------------------- +# Error classification (AC #7) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "status_code,expected_type", + [ + (401, UserAuthError), # invalid / expired / revoked token -> reconnect required + (403, UserAuthError), # missing scope / insufficient permission -> reconnect required + (429, RateLimitError), # rate limited + (404, UserError), # generic client error + (422, UserError), + (500, ProviderError), # exactly 500 must classify (regression: was `> 500`) + (502, ProviderError), + ], +) +def test_error_for_status_classification(status_code, expected_type): + error = _connection_config()._error_for_status(status_code, "boom") + assert isinstance(error, expected_type) + + +def test_error_for_status_returns_none_when_unhandled(): + assert _connection_config()._error_for_status(200, "ok") is None + + +def test_wrap_http_error_maps_forbidden_to_auth_error(): + fake = SimpleNamespace(response=SimpleNamespace(status_code=403, text="requires scope repo")) + error = _connection_config().wrap_http_error(fake) + assert isinstance(error, UserAuthError) + + +def test_wrap_http_error_unhandled_falls_back(): + fake = SimpleNamespace(response=SimpleNamespace(status_code=302, text="redirect")) + error = _connection_config().wrap_http_error(fake) + assert isinstance(error, UnstructuredIngestError) + + +def _patch_github_module(): + """Inject a fake ``github``/``github.GithubException`` so tests don't require the + optional pygithub extra. Returns the patcher and the fake exception module.""" + github_mod = types.ModuleType("github") + exc_mod = types.ModuleType("github.GithubException") + + class GithubException(Exception): + def __init__(self, status=None, data=None, headers=None): + self.status = status + self.data = data if data is not None else {} + self.headers = headers + super().__init__(str(data)) + + class RateLimitExceededException(GithubException): + pass + + class UnknownObjectException(GithubException): + pass + + exc_mod.GithubException = GithubException + exc_mod.RateLimitExceededException = RateLimitExceededException + exc_mod.UnknownObjectException = UnknownObjectException + github_mod.GithubException = exc_mod + patcher = mock.patch.dict( + sys.modules, {"github": github_mod, "github.GithubException": exc_mod} + ) + return patcher, exc_mod + + +def test_wrap_github_exception_unauthorized(): + patcher, _ = _patch_github_module() + with patcher: + e = SimpleNamespace(data={"message": "Bad credentials"}, status=401) + error = _connection_config().wrap_github_exception(e) + assert isinstance(error, UserAuthError) + # The provider's free-form error body must not leak into the raised message; only + # the sanitized summary (type + status) is surfaced. + assert "Bad credentials" not in str(error) + assert "401" in str(error) + + +def test_wrap_github_exception_rate_limit_by_type(): + patcher, exc = _patch_github_module() + with patcher: + e = exc.RateLimitExceededException() + e.data = {"message": "API rate limit exceeded"} + e.status = 403 # GitHub returns 403 for primary rate limits + error = _connection_config().wrap_github_exception(e) + assert isinstance(error, RateLimitError) + + +def test_wrap_github_exception_forbidden_is_auth_error(): + patcher, _ = _patch_github_module() + with patcher: + e = SimpleNamespace(data={"message": "Resource not accessible"}, status=403) + error = _connection_config().wrap_github_exception(e) + assert isinstance(error, UserAuthError) + + +def test_wrap_github_exception_unhandled_returns_original(): + """An unclassified status (e.g. 302) returns the original exception unchanged.""" + patcher, exc = _patch_github_module() + with patcher: + e = exc.GithubException(status=302, data={"message": "redirect"}) + wrapped = _connection_config().wrap_github_exception(e) + assert wrapped is e + + +def test_wrap_github_exception_non_dict_data(): + """Non-dict e.data must not break classification (the body is not read anymore).""" + patcher, exc = _patch_github_module() + with patcher: + e = exc.GithubException(status=401, data="not-a-dict") + wrapped = _connection_config().wrap_github_exception(e) + assert isinstance(wrapped, UserAuthError) + + +def test_wrap_error_dispatches_github_exception(): + patcher, exc = _patch_github_module() + with patcher: + e = exc.GithubException(status=401, data={"message": "Bad creds"}) + wrapped = _connection_config().wrap_error(e) + assert isinstance(wrapped, UserAuthError) + + +def test_wrap_error_dispatches_http_error(): + import requests + + patcher, _ = _patch_github_module() + e = requests.HTTPError(response=SimpleNamespace(status_code=404, text="nope")) + with patcher: + wrapped = _connection_config().wrap_error(e) + assert isinstance(wrapped, UserError) + + +def test_wrap_error_generic_returns_ingest_error(): + patcher, _ = _patch_github_module() + with patcher: + wrapped = _connection_config().wrap_error(RuntimeError("weird")) + assert isinstance(wrapped, UnstructuredIngestError) + + +# --------------------------------------------------------------------------- +# Indexer: version, listing, truncation, and API-call economy +# --------------------------------------------------------------------------- + + +def _indexer(**index_kwargs) -> GithubIndexer: + return GithubIndexer( + connection_config=_connection_config(), + index_config=GithubIndexerConfig(**index_kwargs), + ) + + +def _entry(path, type="blob", sha="sha", size=1, mode="100644", url="u"): + return SimpleNamespace(path=path, type=type, sha=sha, size=size, mode=mode, url=url) + + +class _FakeTree: + def __init__(self, entries, truncated=False): + self.tree = entries + self.truncated = truncated + + +class _FakeRepo: + """Records get_git_tree calls and serves trees keyed by tree-ish.""" + + def __init__(self, trees, default_branch="main"): + self._trees = trees + self.default_branch = default_branch + self.get_git_tree_calls = [] + + def get_git_tree(self, tree_ish, recursive=False): + self.get_git_tree_calls.append((tree_ish, recursive)) + return self._trees[tree_ish] + + +def test_convert_element_maps_git_file_to_file_data(): + """convert_element uses the blob SHA as version and carries size/mode/url through.""" + indexer = _indexer() + git_file = _GitFile( + path="src/app.py", + sha="deadbeefdeadbeefdeadbeefdeadbeefdeadbeef", + size=123, + mode="100644", + url="https://api.github.com/repos/dcneiner/Downloadify/git/blobs/deadbeef", + ) + + file_data = indexer.convert_element(git_file, "main") + + assert file_data.metadata.version == git_file.sha + assert file_data.metadata.filesize_bytes == 123 + assert file_data.metadata.date_modified is None # no reliable per-file source + assert file_data.source_identifiers.relative_path == "src/app.py" + assert file_data.source_identifiers.fullpath == "main/src/app.py" + assert file_data.display_name.endswith("/blob/main/src/app.py") + + +def test_list_files_fast_path_skips_empties_dirs_and_submodules(): + repo = _FakeRepo( + { + "main": _FakeTree( + [ + _entry("README.md", size=10), + _entry("empty.txt", size=0), # empty file -> skipped + _entry("src", type="tree", size=None), # directory -> skipped + _entry("vendored", type="commit", size=None), # submodule -> skipped + _entry("src/app.py", size=42), + ] + ) + } + ) + indexer = _indexer() + + files = indexer.list_files(repo, "main") + + assert sorted(f.path for f in files) == ["README.md", "src/app.py"] + # Fast path is a single recursive call. + assert repo.get_git_tree_calls == [("main", True)] + + +def test_list_files_falls_back_to_per_directory_walk_when_truncated(): + repo = _FakeRepo( + { + # Recursive call truncates -> triggers the walk. + "main": _FakeTree([], truncated=True), + # Walk: root is fetched non-recursively, then each subtree by sha. + "main_nonrecursive": _FakeTree( + [_entry("README.md", size=10), _entry("src", type="tree", sha="src-sha", size=None)] + ), + "src-sha": _FakeTree([_entry("app.py", size=42, sha="app-sha")]), + } + ) + + # get_git_tree is called with the same tree-ish "main" for both the recursive + # probe and the non-recursive root walk; disambiguate by the recursive flag. + def get_git_tree(tree_ish, recursive=False): + repo.get_git_tree_calls.append((tree_ish, recursive)) + if tree_ish == "main": + return repo._trees["main"] if recursive else repo._trees["main_nonrecursive"] + return repo._trees[tree_ish] + + repo.get_git_tree = get_git_tree + indexer = _indexer() + + files = indexer.list_files(repo, "main") + + assert sorted(f.path for f in files) == ["README.md", "src/app.py"] + assert ("main", True) in repo.get_git_tree_calls # recursive probe + assert ("main", False) in repo.get_git_tree_calls # root walk + assert ("src-sha", False) in repo.get_git_tree_calls # subtree walk + + +def test_run_fetches_repo_only_once(monkeypatch): + """Regression guard: convert_element must not re-fetch the repo per file.""" + repo = _FakeRepo( + {"main": _FakeTree([_entry("a.py", size=1, sha="s1"), _entry("b.py", size=1, sha="s2")])} + ) + calls = {"get_repo": 0} + + def fake_get_repo(self): + calls["get_repo"] += 1 + return repo + + monkeypatch.setattr(GithubConnectionConfig, "get_repo", fake_get_repo) + indexer = _indexer() + + file_data = list(indexer.run()) + + assert calls["get_repo"] == 1 # not 2N + 1 + assert repo.get_git_tree_calls == [("main", True)] + assert {fd.metadata.version for fd in file_data} == {"s1", "s2"} + + +def test_get_branch_prefers_pinned_branch_over_default(): + repo = _FakeRepo({}, default_branch="main") + assert _indexer().get_branch(repo) == "main" + assert _indexer(branch="develop").get_branch(repo) == "develop" + + +def test_convert_element_identifier_is_stable_uuid5(): + """The record identifier must be a deterministic uuid5 of the blob URL so the + platform can match records across runs (incremental skip depends on it).""" + from uuid import NAMESPACE_DNS, uuid5 + + indexer = _indexer() + git_file = _GitFile(path="src/app.py", sha="sha", size=1, mode="100644", url="u") + + file_data = indexer.convert_element(git_file, "main") + + expected_url = "https://github.com/dcneiner/Downloadify/blob/main/src/app.py" + assert file_data.identifier == str(uuid5(NAMESPACE_DNS, expected_url)) + # Stable across repeated conversions of the same file. + assert indexer.convert_element(git_file, "main").identifier == file_data.identifier + assert file_data.metadata.permissions_data == [{"mode": "100644"}] + + +def test_list_files_honors_recursive_false(): + repo = _FakeRepo({"main": _FakeTree([_entry("README.md", size=1)])}) + + files = _indexer(recursive=False).list_files(repo, "main") + + assert [f.path for f in files] == ["README.md"] + assert repo.get_git_tree_calls == [("main", False)] + + +def test_list_files_recursive_false_truncated_returns_partial(): + """A truncated non-recursive tree can't be paginated or descended into, so the + connector returns the partial listing (with a warning) rather than walking.""" + repo = _FakeRepo({"main": _FakeTree([_entry("a.txt", size=1)], truncated=True)}) + + files = _indexer(recursive=False).list_files(repo, "main") + + assert [f.path for f in files] == ["a.txt"] + assert repo.get_git_tree_calls == [("main", False)] # no walk + + +def test_walk_tree_reconstructs_nested_paths(): + repo = _FakeRepo( + { + "root-sha": _FakeTree( + [ + _entry("a", type="tree", sha="a-sha", size=None), + _entry("top.txt", size=1), + ] + ), + "a-sha": _FakeTree( + [ + _entry("b", type="tree", sha="b-sha", size=None), + _entry("mid.txt", size=1), + ] + ), + "b-sha": _FakeTree([_entry("deep.txt", size=1)]), + } + ) + + files = _indexer()._walk_tree(repo, "root-sha", prefix="") + + assert sorted(f.path for f in files) == ["a/b/deep.txt", "a/mid.txt", "top.txt"] + + +def test_precheck_succeeds_when_repo_reachable(monkeypatch): + monkeypatch.setattr(GithubConnectionConfig, "get_repo", lambda self: object()) + _indexer().precheck() # must not raise + + +def test_precheck_wraps_connection_failure(monkeypatch): + patcher, _ = _patch_github_module() + + def boom(self): + raise RuntimeError("cannot reach github") + + monkeypatch.setattr(GithubConnectionConfig, "get_repo", boom) + with patcher, pytest.raises(UnstructuredIngestError): + _indexer().precheck() + + +def test_conform_url_keeps_owner_repo_pair(): + conn = _connection_config() + assert conn.url == "dcneiner/Downloadify" + assert conn.get_full_url() == "https://github.com/dcneiner/Downloadify" + + +# --------------------------------------------------------------------------- +# Downloader: repo caching, fetch, content resolution, and file write +# --------------------------------------------------------------------------- + + +def _downloader(**download_kwargs) -> GithubDownloader: + return GithubDownloader( + download_config=GithubDownloaderConfig(**download_kwargs), + connection_config=_connection_config(), + ) + + +def _file_data(path: str = "README.md") -> FileData: + return _indexer().convert_element( + _GitFile(path=path, sha="sha", size=1, mode="100644", url="u"), "main" + ) + + +def test_downloader_caches_repo(monkeypatch): + """_get_repo memoizes so repeated get_file calls don't re-issue GET /repos.""" + calls = {"n": 0} + sentinel = object() + + def fake_get_repo(self): + calls["n"] += 1 + return sentinel + + monkeypatch.setattr(GithubConnectionConfig, "get_repo", fake_get_repo) + downloader = _downloader() + + assert downloader._get_repo() is sentinel + assert downloader._get_repo() is sentinel + assert calls["n"] == 1 + + +def test_get_file_returns_content_file(): + patcher, _ = _patch_github_module() + downloader = _downloader() + sentinel = object() + downloader._repo = SimpleNamespace(get_contents=lambda path: sentinel) + with patcher: + assert downloader.get_file(_file_data("README.md")) is sentinel + + +def test_get_file_missing_raises_user_error(): + patcher, exc = _patch_github_module() + downloader = _downloader() + + def boom(path): + raise exc.UnknownObjectException(status=404, data={"message": "Not Found"}) + + downloader._repo = SimpleNamespace(get_contents=boom) + with patcher, pytest.raises(UserError): + downloader.get_file(_file_data("missing.txt")) + + +def test_get_contents_returns_decoded_content(): + """Base64-decoded content from the contents API is used without a second fetch.""" + downloader = _downloader() + content_file = SimpleNamespace(decoded_content=b"hello", download_url="https://raw/x") + assert downloader.get_contents(content_file) == b"hello" + + +def test_get_contents_falls_back_to_download_url(monkeypatch): + """Large files return empty decoded_content, so we fetch the raw download URL.""" + downloader = _downloader() + content_file = SimpleNamespace(decoded_content=b"", download_url="https://raw/big") + captured = {} + + def fake_get(url, **kwargs): + captured["url"] = url + captured["timeout"] = kwargs.get("timeout") + return SimpleNamespace(raise_for_status=lambda: None, content=b"raw-bytes") + + monkeypatch.setattr("requests.get", fake_get) + + assert downloader.get_contents(content_file) == b"raw-bytes" + assert captured["url"] == "https://raw/big" + # The raw-blob fetch must be bounded so a hung connection can't stall a worker. + assert captured["timeout"] is not None + + +def test_get_contents_wraps_http_error(monkeypatch): + import requests + + patcher, _ = _patch_github_module() + # max_retries=1: a 500 is now a retriable ProviderError, so a single attempt keeps + # the test fast while still asserting the HTTPError is wrapped, not leaked raw. + downloader = _downloader(max_retries=1) + content_file = SimpleNamespace(decoded_content=None, download_url="https://raw/big") + + def raise_for_status(): + raise requests.HTTPError(response=SimpleNamespace(status_code=500, text="boom")) + + monkeypatch.setattr( + "requests.get", + lambda url, **kwargs: SimpleNamespace(raise_for_status=raise_for_status, content=b""), + ) + with patcher, pytest.raises(ProviderError): + downloader.get_contents(content_file) + + +def test_downloader_run_writes_file(tmp_path): + patcher, _ = _patch_github_module() + downloader = _downloader(download_dir=tmp_path) + downloader._repo = SimpleNamespace( + get_contents=lambda path: SimpleNamespace(decoded_content=b"payload", download_url=None) + ) + file_data = _file_data("README.md") + + with patcher: + result = downloader.run(file_data) + + written = tmp_path / "README.md" + assert written.read_bytes() == b"payload" + assert result["path"] == written + assert file_data.local_download_path == str(written.resolve()) + + +def test_downloader_run_creates_parent_directories(tmp_path): + # Regression guard: run() must mkdir the full parent chain before writing, or + # files in subdirectories (and the download dir itself) fail with FileNotFoundError. + patcher, _ = _patch_github_module() + downloader = _downloader(download_dir=tmp_path) + downloader._repo = SimpleNamespace( + get_contents=lambda path: SimpleNamespace(decoded_content=b"x", download_url=None) + ) + file_data = _file_data("src/app.py") + + with patcher: + downloader.run(file_data) + + assert (tmp_path / "src" / "app.py").read_bytes() == b"x" + + +# --------------------------------------------------------------------------- +# Rate limiting & retry: backoff parsing, error classification, retry loop +# --------------------------------------------------------------------------- + + +@pytest.fixture +def no_sleep(monkeypatch): + """Make tenacity's backoff instant so retry-loop tests don't actually sleep. + + tenacity's default nap looks up ``time.sleep`` at call time, so patching the time + module (rather than the bound-at-import ``tenacity.nap.sleep``) is what takes effect. + """ + monkeypatch.setattr("time.sleep", lambda *_a, **_k: None) + + +def test_parse_retry_after_delta_seconds(): + assert _parse_retry_after({"Retry-After": "42"}) == 42.0 + # Header lookup is case-insensitive (PyGithub vs requests casing). + assert _parse_retry_after({"retry-after": "7"}) == 7.0 + assert _parse_retry_after({}) is None + assert _parse_retry_after({"Retry-After": "not-a-number"}) is None + + +def test_parse_retry_after_http_date(): + # An HTTP-date ~30s in the future should resolve to roughly 30s of wait. + from datetime import datetime, timedelta, timezone + from email.utils import format_datetime + + future = datetime.now(timezone.utc) + timedelta(seconds=30) + seconds = _parse_retry_after({"Retry-After": format_datetime(future)}) + assert seconds is not None + assert 20 <= seconds <= 30 + + +def test_rate_limit_backoff_prefers_retry_after_over_reset(): + reset = str(int(time()) + 999) + headers = {"Retry-After": "5", "x-ratelimit-remaining": "0", "x-ratelimit-reset": reset} + assert _rate_limit_backoff(headers) == 5.0 + + +def test_rate_limit_backoff_uses_reset_when_window_exhausted(): + reset = str(int(time()) + 45) + backoff = _rate_limit_backoff({"x-ratelimit-remaining": "0", "x-ratelimit-reset": reset}) + assert backoff is not None + assert 40 <= backoff <= 45 + + +def test_rate_limit_backoff_none_when_window_not_exhausted(): + # Remaining budget left -> no primary-limit backoff hint. + assert _rate_limit_backoff({"x-ratelimit-remaining": "17"}) is None + assert _rate_limit_backoff({}) is None + + +def test_honor_retry_after_caps_at_max(): + err = RateLimitError("throttled") + err.retry_after = 10_000 # absurdly long window + assert _honor_retry_after(base_seconds=2.0, exc=err) == _MAX_RETRY_AFTER_WAIT + + +def test_honor_retry_after_falls_back_to_base_when_absent(): + assert _honor_retry_after(base_seconds=3.0, exc=RuntimeError("no hint")) == 3.0 + + +def test_wrap_github_exception_stamps_retry_after_from_header(): + patcher, exc = _patch_github_module() + with patcher: + e = exc.RateLimitExceededException( + status=403, data={"message": "rate limited"}, headers={"Retry-After": "30"} + ) + err = _connection_config().wrap_github_exception(e) + assert isinstance(err, RateLimitError) + assert err.retry_after == 30.0 + + +def test_wrap_github_exception_stamps_retry_after_from_reset(): + patcher, exc = _patch_github_module() + reset = str(int(time()) + 45) + with patcher: + e = exc.RateLimitExceededException( + status=403, + data={"message": "rl"}, + headers={"x-ratelimit-remaining": "0", "x-ratelimit-reset": reset}, + ) + err = _connection_config().wrap_github_exception(e) + assert isinstance(err, RateLimitError) + assert 40 <= err.retry_after <= 45 + + +def test_wrap_http_error_403_rate_limit_is_rate_limit_error(): + reset = str(int(time()) + 30) + resp = SimpleNamespace( + status_code=403, + text="rate limited", + headers={"x-ratelimit-remaining": "0", "x-ratelimit-reset": reset}, + ) + err = _connection_config().wrap_http_error(SimpleNamespace(response=resp)) + assert isinstance(err, RateLimitError) + assert 0 < err.retry_after <= 30 + + +def test_wrap_http_error_429_is_rate_limit_error(): + resp = SimpleNamespace(status_code=429, text="slow down", headers={"Retry-After": "12"}) + err = _connection_config().wrap_http_error(SimpleNamespace(response=resp)) + assert isinstance(err, RateLimitError) + assert err.retry_after == 12.0 + + +def test_wrap_http_error_secondary_rate_limit_403_is_rate_limit_error(): + # Secondary/abuse limit: GitHub sends 403 + Retry-After while the PRIMARY budget + # (x-ratelimit-remaining) is still positive. Must be a retriable RateLimitError, + # not a permanent UserAuthError. + resp = SimpleNamespace( + status_code=403, + text="You have exceeded a secondary rate limit", + headers={"Retry-After": "45", "x-ratelimit-remaining": "4999"}, + ) + err = _connection_config().wrap_http_error(SimpleNamespace(response=resp)) + assert isinstance(err, RateLimitError) + assert err.retry_after == 45.0 + + +def test_wrap_github_exception_secondary_rate_limit_403_is_rate_limit_error(): + # Same secondary-limit case via PyGithub, but as a plain GithubException (not typed + # RateLimitExceededException) — header detection must still classify it as a throttle. + patcher, exc = _patch_github_module() + with patcher: + e = exc.GithubException( + status=403, + data={"message": "You have exceeded a secondary rate limit"}, + headers={"Retry-After": "60", "x-ratelimit-remaining": "4321"}, + ) + err = _connection_config().wrap_github_exception(e) + assert isinstance(err, RateLimitError) + assert err.retry_after == 60.0 + + +def test_wrap_http_error_does_not_leak_response_body(): + # The raw upstream response body must never reach the raised error message. + secret = "ghs_supersecrettoken_should_never_surface" + resp = SimpleNamespace(status_code=500, text=f"internal error {secret}", headers={}) + err = _connection_config().wrap_http_error(SimpleNamespace(response=resp)) + assert isinstance(err, ProviderError) + assert secret not in str(err) + + +def test_wrap_error_maps_network_error_to_retriable(): + import requests + + patcher, _ = _patch_github_module() + with patcher: + wrapped = _connection_config().wrap_error(requests.ConnectionError("reset by peer")) + assert isinstance(wrapped, SourceConnectionNetworkError) + + +def test_run_with_retry_retries_then_succeeds(no_sleep): + conn = _connection_config() + calls = {"n": 0} + + def fn(): + calls["n"] += 1 + if calls["n"] < 3: + raise RateLimitError("throttled") + return "ok" + + assert conn.run_with_retry(fn, max_retries=5) == "ok" + assert calls["n"] == 3 + + +def test_run_with_retry_does_not_retry_user_error(no_sleep): + conn = _connection_config() + calls = {"n": 0} + + def fn(): + calls["n"] += 1 + raise UserAuthError("bad token") + + with pytest.raises(UserAuthError): + conn.run_with_retry(fn, max_retries=5) + assert calls["n"] == 1 # auth failures are permanent — no retry + + +def test_run_with_retry_exhausts_and_reraises(no_sleep): + conn = _connection_config() + calls = {"n": 0} + + def fn(): + calls["n"] += 1 + raise ProviderError("upstream 500") + + with pytest.raises(ProviderError): + conn.run_with_retry(fn, max_retries=4) + assert calls["n"] == 4 + + +def test_run_with_retry_wraps_and_retries_raw_network_error(no_sleep): + import requests + + patcher, _ = _patch_github_module() + conn = _connection_config() + calls = {"n": 0} + + def fn(): + calls["n"] += 1 + raise requests.ConnectionError("reset by peer") + + with patcher, pytest.raises(SourceConnectionNetworkError): + conn.run_with_retry(fn, max_retries=3) + assert calls["n"] == 3 # network blips are transient -> retried then reraised + + +def test_run_with_retry_honors_stamped_retry_after(monkeypatch): + """The retry wait uses the error's stamped retry_after when it exceeds the base.""" + conn = _connection_config() + seen_waits = [] + monkeypatch.setattr("time.sleep", lambda seconds, *_a, **_k: seen_waits.append(seconds)) + + calls = {"n": 0} + + def fn(): + calls["n"] += 1 + if calls["n"] < 2: + err = RateLimitError("throttled") + err.retry_after = 25 + raise err + return "done" + + assert conn.run_with_retry(fn, max_retries=5) == "done" + assert seen_waits and seen_waits[0] == 25 diff --git a/unstructured_ingest/__version__.py b/unstructured_ingest/__version__.py index be7f33e10..0f6c00010 100644 --- a/unstructured_ingest/__version__.py +++ b/unstructured_ingest/__version__.py @@ -1 +1 @@ -__version__ = "1.7.3" # pragma: no cover +__version__ = "1.7.4" # pragma: no cover diff --git a/unstructured_ingest/processes/connectors/github.py b/unstructured_ingest/processes/connectors/github.py index 4aac96587..d624a127f 100644 --- a/unstructured_ingest/processes/connectors/github.py +++ b/unstructured_ingest/processes/connectors/github.py @@ -1,7 +1,8 @@ +import builtins from dataclasses import dataclass, field from pathlib import Path from time import time -from typing import TYPE_CHECKING, Any, Generator, Optional +from typing import TYPE_CHECKING, Any, Callable, Generator, Optional from urllib.parse import urlparse from uuid import NAMESPACE_DNS, uuid5 @@ -14,9 +15,12 @@ ) from unstructured_ingest.error import ( ProviderError, + RateLimitError, + SourceConnectionNetworkError, UnstructuredIngestError, UserAuthError, UserError, + ValueError, safe_error_summary, ) from unstructured_ingest.interfaces import ( @@ -42,9 +46,113 @@ CONNECTOR_TYPE = "github" +# Cap on honored backoff: GitHub's primary-limit reset can be an hour out, and we +# won't stall a run that long on one Retry-After / X-RateLimit-Reset. +_MAX_RETRY_AFTER_WAIT = 300.0 + + +def _header_value(headers: Any, name: str) -> Optional[str]: + """Read a header case-insensitively (PyGithub and requests differ on key casing).""" + if not headers: + return None + try: + if name in headers: + return headers[name] + lname = name.lower() + for key, value in headers.items(): + if key.lower() == lname: + return value + except (builtins.TypeError, builtins.AttributeError): + return None + return None + + +def _parse_retry_after(headers: Any) -> Optional[float]: + """Parse a ``Retry-After`` header (delta-seconds or HTTP-date) into seconds.""" + value = _header_value(headers, "Retry-After") + if not value: + return None + # This module shadows the builtin ``ValueError``, so catch builtins explicitly. + try: + return max(0.0, float(value)) + except (builtins.TypeError, builtins.ValueError): + pass + try: + from datetime import datetime, timezone + from email.utils import parsedate_to_datetime + + dt = parsedate_to_datetime(value) + if dt is not None: + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return max(0.0, (dt - datetime.now(timezone.utc)).total_seconds()) + except (builtins.TypeError, builtins.ValueError): + pass + return None + + +def _rate_limit_backoff(headers: Any) -> Optional[float]: + """Seconds to wait before retrying a throttle: prefer the secondary-limit + ``Retry-After``, else the primary-limit ``x-ratelimit-reset`` once the window is + exhausted (``x-ratelimit-remaining: 0``). ``None`` when no hint is available.""" + retry_after = _parse_retry_after(headers) + if retry_after is not None: + return retry_after + if str(_header_value(headers, "x-ratelimit-remaining")) == "0": + reset = _header_value(headers, "x-ratelimit-reset") + if reset is not None: + try: + return max(0.0, float(reset) - time()) + except (builtins.TypeError, builtins.ValueError): + return None + return None + + +def _is_rate_limited(status_code: Optional[int], headers: Any) -> bool: + """True when a GitHub response is throttled. 429 is always a throttle; a 403 is a + throttle when the primary window is exhausted (``x-ratelimit-remaining: 0``) OR a + ``Retry-After`` header is present — secondary/abuse limits set ``Retry-After`` even + while the primary budget is still positive, and those must be retried, not treated + as a permanent auth failure.""" + if status_code == 429: + return True + if status_code == 403: + if _header_value(headers, "Retry-After") is not None: + return True + if str(_header_value(headers, "x-ratelimit-remaining")) == "0": + return True + return False + + +def _honor_retry_after(base_seconds: float, exc: BaseException) -> float: + """Use the server's stamped ``retry_after`` when it exceeds the backoff, capped.""" + retry_after = getattr(exc, "retry_after", None) + if isinstance(retry_after, (int, float)) and retry_after > base_seconds: + return min(float(retry_after), _MAX_RETRY_AFTER_WAIT) + return base_seconds + + +def _should_retry(exc: BaseException) -> bool: + """Retry throttles, transient 5xx, and network blips — never auth/user errors.""" + return isinstance(exc, (RateLimitError, ProviderError, SourceConnectionNetworkError)) + class GithubAccessConfig(AccessConfig): - access_token: str = Field(description="Github acess token") + access_token: Optional[str] = Field( + description="Github personal access token (PAT)", + default=None, + ) + oauth_token: Optional[str] = Field( + description="GitHub OAuth 2.0 access token", + default=None, + ) + refresh_token: Optional[str] = Field( + description=( + "GitHub OAuth 2.0 refresh token. Used by the platform to refresh expired " + "access tokens before each job run; the connector never uses it directly." + ), + default=None, + ) class GithubConnectionConfig(ConnectionConfig): @@ -56,6 +164,17 @@ def conform_url(cls, value: str): parsed_url = urlparse(value) return parsed_url.path + def model_post_init(self, __context): + access_configs = self.access_config.get_secret_value() + pat_auth = bool(access_configs.access_token) + oauth_auth = bool(access_configs.oauth_token) + if pat_auth and oauth_auth: + raise ValueError( + "access_token (PAT) and oauth_token are mutually exclusive, provide only one" + ) + if not (pat_auth or oauth_auth): + raise ValueError("no form of auth provided, set access_token or oauth_token") + def get_full_url(self): return f"https://github.com/{self.url}" @@ -63,48 +182,152 @@ def get_full_url(self): def get_client(self) -> "GithubClient": from github import Github as GithubClient - return GithubClient(login_or_token=self.access_config.get_secret_value().access_token) + access_configs = self.access_config.get_secret_value() + # The platform's init-oauth-refresh container populates oauth_token with a fresh + # access token before the job runs (per PLU-381); refresh is not handled here. + token = access_configs.oauth_token or access_configs.access_token + return GithubClient(login_or_token=token) def get_repo(self) -> "Repository": client = self.get_client() return client.get_repo(self.url) - def wrap_github_exception(self, e: "GithubException") -> Exception: - data = e.data - status_code = e.status - message = data.get("message") + def _error_for_status(self, status_code: int, message: str) -> Optional[Exception]: + """Map a GitHub HTTP status to a classified error, or None if unhandled. + + 401 (invalid/expired/revoked token) and 403 (missing scope / insufficient + permission) both map to UserAuthError so the platform prompts a reconnect. + """ + if status_code == 429: + return RateLimitError(f"GitHub API rate limit exceeded: {message}") if status_code == 401: return UserAuthError(f"Unauthorized access to Github: {message}") + if status_code == 403: + return UserAuthError( + f"Forbidden access to Github (missing scope or insufficient permission): {message}" + ) if 400 <= status_code < 500: return UserError(message) - if status_code > 500: + if status_code >= 500: return ProviderError(message) - logger.debug(f"unhandled github error: {safe_error_summary(e)}") - return e + return None + + def wrap_github_exception(self, e: "GithubException") -> Exception: + from github.GithubException import RateLimitExceededException + + headers = getattr(e, "headers", None) or {} + # Never interpolate the provider's free-form error body into raised/logged + # messages — it can carry request data or secrets. The sanitized summary + # (type + allowlisted status/fields) is enough to act on. + summary = safe_error_summary(e) + # PyGithub types primary rate limits as RateLimitExceededException, but relies on + # message-text matching for secondary/abuse limits, so also detect a throttle from + # the response headers (Retry-After / exhausted window). + if isinstance(e, RateLimitExceededException) or _is_rate_limited( + getattr(e, "status", None), headers + ): + error: Optional[Exception] = RateLimitError( + f"GitHub API rate limit exceeded: {summary}" + ) + else: + error = self._error_for_status(e.status, summary) + if error is None: + logger.debug(f"unhandled github error: {summary}") + return e + # Stamp the throttle backoff so a retry can honor GitHub's requested window. + if isinstance(error, RateLimitError): + backoff = _rate_limit_backoff(headers) + if backoff is not None: + error.retry_after = backoff + return error def wrap_http_error(self, e: "HTTPError") -> Exception: - status_code = e.response.status_code - if status_code == 401: - return UserAuthError(f"Unauthorized access to Github: {e.response.text}") - if 400 <= status_code < 500: - return UserError(e.response.text) - if status_code > 500: - return ProviderError(e.response.text) - logger.debug(f"unhandled http error: {safe_error_summary(e)}") - return UnstructuredIngestError(safe_error_summary(e)) + response = e.response + headers = getattr(response, "headers", None) or {} + status_code = response.status_code + # Sanitized summary only — never the raw (unbounded) response body. + summary = safe_error_summary(e) + # 429, or a 403 that is a primary (exhausted window) or secondary/abuse + # (Retry-After present) throttle, is a retriable RateLimitError, not an auth error. + if _is_rate_limited(status_code, headers): + error = RateLimitError(f"GitHub API rate limit exceeded: {summary}") + backoff = _rate_limit_backoff(headers) + if backoff is not None: + error.retry_after = backoff + return error + mapped = self._error_for_status(status_code, summary) + if mapped is not None: + return mapped + logger.debug(f"unhandled http error: {summary}") + return UnstructuredIngestError(summary) @requires_dependencies(["requests"], extras="github") def wrap_error(self, e: Exception) -> Exception: from github.GithubException import GithubException - from requests import HTTPError + from requests import ConnectionError as RequestsConnectionError + from requests import HTTPError, Timeout if isinstance(e, GithubException): return self.wrap_github_exception(e=e) if isinstance(e, HTTPError): return self.wrap_http_error(e=e) + # A connection reset / timeout contacting GitHub is transient — classify it as a + # retriable network error rather than an opaque failure. + if isinstance(e, (RequestsConnectionError, Timeout)): + return SourceConnectionNetworkError( + f"Network error contacting GitHub: {safe_error_summary(e)}" + ) logger.debug(f"unhandled error: {safe_error_summary(e)}") return UnstructuredIngestError(safe_error_summary(e)) + @requires_dependencies(["tenacity"], extras="github") + def run_with_retry(self, fn: Callable[[], Any], *, max_retries: int) -> Any: + """Run *fn* (a GitHub call), retrying transient throttles / 5xx / network errors + with exponential backoff that honors GitHub's ``Retry-After``. Failures are + normalized through ``wrap_error`` so the predicate sees our typed errors.""" + from tenacity import ( + retry, + retry_if_exception, + stop_after_attempt, + wait_exponential, + ) + + exp_wait = wait_exponential(exp_base=2, multiplier=1, min=2, max=30) + + def _wait(retry_state) -> float: + base = exp_wait(retry_state) + exc = retry_state.outcome.exception() if retry_state.outcome else None + return _honor_retry_after(base, exc) if exc is not None else base + + @retry( + stop=stop_after_attempt(max_retries), + wait=_wait, + retry=retry_if_exception(_should_retry), + reraise=True, + ) + def _wrapped() -> Any: + try: + return fn() + except UnstructuredIngestError: + # Already typed (retry_after preserved) — don't re-wrap and lose it. + raise + except Exception as e: + raise self.wrap_error(e=e) from e + + return _wrapped() + + +@dataclass +class _GitFile: + """A blob entry resolved to its full repo-relative path, so ``convert_element`` + sees one uniform shape from both the recursive-tree and truncation-walk paths.""" + + path: str + sha: str + size: int + mode: str + url: str + class GithubIndexerConfig(IndexerConfig): branch: Optional[str] = Field( @@ -113,6 +336,11 @@ class GithubIndexerConfig(IndexerConfig): recursive: bool = Field( description="Recursively index all files in the repository", default=True ) + max_retries: int = Field( + description="Max attempts for a GitHub API call before giving up on a " + "throttle / transient 5xx / network error", + default=10, + ) @dataclass @@ -122,57 +350,122 @@ class GithubIndexer(Indexer): connector_type: str = CONNECTOR_TYPE def precheck(self) -> None: - try: - self.connection_config.get_repo() - except Exception as e: - raise self.connection_config.wrap_error(e=e) - - def get_branch(self) -> str: - repo = self.connection_config.get_repo() - sha = self.index_config.branch or repo.default_branch - return sha - - def list_files(self) -> list["GitTreeElement"]: - repo = self.connection_config.get_repo() - sha = self.index_config.branch or repo.default_branch - git_tree = repo.get_git_tree(sha, recursive=self.index_config.recursive) - file_elements = [ - element for element in git_tree.tree if element.size is not None and element.size > 0 - ] - return file_elements - - def convert_element(self, element: "GitTreeElement") -> FileData: - full_path = ( - f"{self.connection_config.get_full_url()}/blob/{self.get_branch()}/{element.path}" + # run_with_retry already normalizes failures to typed errors, so no wrapping here. + self.connection_config.run_with_retry( + self.connection_config.get_repo, max_retries=self.index_config.max_retries ) + def get_branch(self, repo: "Repository") -> str: + # No extra call: a pinned branch short-circuits, otherwise default_branch is + # already populated on the repo object we fetch once in run(). + return self.index_config.branch or repo.default_branch + + def list_files(self, repo: "Repository", branch: str) -> list[_GitFile]: + """List every non-empty blob under *branch*, robust to tree truncation. + + Fast path: a single recursive git-tree call. GitHub caps the recursive tree + at ~100k entries / 7 MB and cannot paginate it, so when the response is + truncated we fall back to a per-directory walk that fetches each subtree + individually and reassembles full paths. + """ + recursive = self.index_config.recursive + git_tree = self.connection_config.run_with_retry( + lambda: repo.get_git_tree(branch, recursive=recursive), + max_retries=self.index_config.max_retries, + ) + if not git_tree.truncated: + return self._blobs(git_tree.tree) + + if not recursive: + # A non-recursive listing can't be completed — git trees have no page + # param and there's no lower level to descend into. + logger.warning( + "GitHub root tree for %s is truncated; some files may be missing", + self.connection_config.url, + ) + return self._blobs(git_tree.tree) + + logger.warning( + "GitHub tree for %s is truncated; falling back to a per-directory walk", + self.connection_config.url, + ) + return self._walk_tree(repo, branch, prefix="") + + @staticmethod + def _blobs(entries: list["GitTreeElement"], prefix: str = "") -> list[_GitFile]: + # Keep only blobs with content: skips trees (directories), submodules + # (type "commit"), and empty files (size 0 / None). + files: list[_GitFile] = [] + for e in entries: + if e.type == "blob" and e.size: + path = f"{prefix}/{e.path}" if prefix else e.path + files.append(_GitFile(path=path, sha=e.sha, size=e.size, mode=e.mode, url=e.url)) + return files + + def _walk_tree(self, repo: "Repository", tree_ish: str, prefix: str) -> list[_GitFile]: + """Collect blobs one directory at a time (recursive-tree truncation fallback).""" + tree = self.connection_config.run_with_retry( + lambda: repo.get_git_tree(tree_ish, recursive=False), + max_retries=self.index_config.max_retries, + ) + if tree.truncated: + # A single directory exceeding the per-tree cap is astronomically rare; + # warn rather than silently drop entries. + logger.warning( + "GitHub subtree %r under %s is truncated; some files may be missing", + prefix or "/", + self.connection_config.url, + ) + files = self._blobs(tree.tree, prefix) + for e in tree.tree: + if e.type == "tree": + child_prefix = f"{prefix}/{e.path}" if prefix else e.path + files.extend(self._walk_tree(repo, e.sha, child_prefix)) + return files + + def convert_element(self, file: _GitFile, branch: str) -> FileData: + full_path = f"{self.connection_config.get_full_url()}/blob/{branch}/{file.path}" + return FileData( identifier=str(uuid5(NAMESPACE_DNS, full_path)), connector_type=self.connector_type, display_name=full_path, source_identifiers=SourceIdentifiers( - filename=Path(element.path).name, - fullpath=(Path(self.get_branch()) / element.path).as_posix(), - rel_path=element.path, + filename=Path(file.path).name, + fullpath=(Path(branch) / file.path).as_posix(), + rel_path=file.path, ), metadata=FileDataSourceMetadata( - url=element.url, - version=element.etag, + url=file.url, + # Blob SHA: a content hash already in the tree listing (no extra call) + # that changes iff content changes — what the platform's per-record + # incremental skip compares on. (The tree ETag is shared across files.) + version=file.sha, record_locator={}, - date_modified=str(element.last_modified_datetime.timestamp()), + # date_modified is intentionally unset: git tree entries carry no + # per-file timestamp, and version is the sole skip signal anyway. date_processed=str(time()), - filesize_bytes=element.size, - permissions_data=[{"mode": element.mode}], + filesize_bytes=file.size, + permissions_data=[{"mode": file.mode}], ), ) def run(self, **kwargs: Any) -> Generator[FileData, None, None]: - for element in self.list_files(): - yield self.convert_element(element=element) + # Fetch the repo (and default_branch) once per run, not per file. + repo = self.connection_config.run_with_retry( + self.connection_config.get_repo, max_retries=self.index_config.max_retries + ) + branch = self.get_branch(repo) + for file in self.list_files(repo, branch): + yield self.convert_element(file, branch) class GithubDownloaderConfig(DownloaderConfig): - pass + max_retries: int = Field( + description="Max attempts for a GitHub API call before giving up on a " + "throttle / transient 5xx / network error", + default=10, + ) @dataclass @@ -180,20 +473,35 @@ class GithubDownloader(Downloader): download_config: GithubDownloaderConfig connection_config: GithubConnectionConfig connector_type: str = CONNECTOR_TYPE + _repo: Optional["Repository"] = field(default=None, init=False, repr=False) + + def _get_repo(self) -> "Repository": + # Cache the repo so repeated get_file calls in one downloader process don't + # each re-issue GET /repos/{owner}/{repo}. + if self._repo is None: + self._repo = self.connection_config.run_with_retry( + self.connection_config.get_repo, max_retries=self.download_config.max_retries + ) + return self._repo @requires_dependencies(["github"], extras="github") def get_file(self, file_data: FileData) -> "ContentFile": from github.GithubException import UnknownObjectException path = file_data.source_identifiers.relative_path - repo = self.connection_config.get_repo() - - try: - content_file = repo.get_contents(path) - except UnknownObjectException as e: - logger.error(f"File doesn't exists {self.connection_config.url}/{path}: {e}") - raise UserError(f"File not found: {path}") - return content_file + repo = self._get_repo() + + def _fetch() -> "ContentFile": + try: + return repo.get_contents(path) + except UnknownObjectException: + # A missing file is a permanent user error — surface it without retrying. + logger.error(f"File doesn't exist: {self.connection_config.url}/{path}") + raise UserError(f"File not found: {path}") + + return self.connection_config.run_with_retry( + _fetch, max_retries=self.download_config.max_retries + ) @requires_dependencies(["requests"], extras="github") def get_contents(self, content_file: "ContentFile") -> bytes: @@ -202,17 +510,27 @@ def get_contents(self, content_file: "ContentFile") -> bytes: if content_file.decoded_content: return content_file.decoded_content download_url = content_file.download_url - resp = requests.get(download_url) - try: - resp.raise_for_status() - except requests.HTTPError as e: - raise self.connection_config.wrap_error(e=e) - return resp.content + + def _fetch() -> bytes: + # Bound the request so a hung raw-blob fetch can't stall a worker. + resp = requests.get(download_url, timeout=60) + try: + resp.raise_for_status() + except requests.HTTPError as e: + raise self.connection_config.wrap_error(e=e) + return resp.content + + return self.connection_config.run_with_retry( + _fetch, max_retries=self.download_config.max_retries + ) def run(self, file_data: FileData, **kwargs: Any) -> download_responses: content_file = self.get_file(file_data) contents = self.get_contents(content_file) download_path = self.get_download_path(file_data) + # Files live under nested paths and the download dir may not exist yet; nothing + # upstream creates the parent chain, so do it here before writing. + download_path.parent.mkdir(parents=True, exist_ok=True) with download_path.open("wb") as f: f.write(contents) return self.generate_download_response(file_data=file_data, download_path=download_path) diff --git a/uv.lock b/uv.lock index e39c2acc8..e720a9107 100644 --- a/uv.lock +++ b/uv.lock @@ -7128,6 +7128,7 @@ gcs = [ github = [ { name = "pygithub" }, { name = "requests" }, + { name = "tenacity" }, ] gitlab = [ { name = "python-gitlab" }, @@ -7470,6 +7471,7 @@ requires-dist = [ { name = "slack-sdk", extras = ["optional"], marker = "extra == 'slack'" }, { name = "snowflake-connector-python", marker = "extra == 'snowflake'" }, { name = "tenacity", marker = "extra == 'delta-table'" }, + { name = "tenacity", marker = "extra == 'github'" }, { name = "tenacity", marker = "extra == 'google-drive'" }, { name = "tenacity", marker = "extra == 'ibm-watsonx-s3'" }, { name = "tenacity", marker = "extra == 'onedrive'" },