Skip to content

Fix the rate limit handling #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mhoarau wants to merge 1 commit into
base: feat/search-pr-closed-before-date
Choose a base branch
from
Open

Fix the rate limit handling #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
# Unreleased
* Retry rate limits using response headers, including HTTP 429. The -1 retry setting now allows ten total attempts.
* Retry optional pull request fields separately from the search; skip unavailable fields after failure.

# Version 1.1.0 - 2026-04-03
* Search GitHub Pull Requests
* Add a `closed_before_date` parameter to restrict closed pull requests by closing date.
Expand Down
2 changes: 1 addition & 1 deletion python-connectors/github_search-pull-requests/connector.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@
{
"name": "number_of_fetch_retry",
"label": "Number of retries:",
"description": "Allow to retry after hitting the rate limit. -1 for no retry limit, 0 to disable.",
"description": "Maximum total fetch attempts when auto retry is enabled. 0 disables retries; -1 allows 10 attempts.",
"type": "INT",
"defaultValue": -1,
"visibilityCondition": "model.enable_auto_retry"
Expand Down
269 changes: 153 additions & 116 deletions python-lib/utils/github_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,15 @@
from github import RateLimitExceededException, GithubException
import logging
import time
from datetime import datetime
from email.utils import parsedate_to_datetime
import math

MAX_NUMBER_OF_FETCH_ATTEMPT_TO_BYPASS_RATE_LIMIT = 2
MAX_NUMBER_OF_FETCH_ATTEMPT_TO_BYPASS_RATE_LIMIT = 10
COSTLY_COMMENTS_FIELD = "comments"
COSTLY_FIELDS = ("merged", "requested_reviewers", "requested_teams", "merged_at",
COSTLY_COMMENTS_FIELD, "base_branch:base.ref", "head_branch:head.ref")
COSTLY_FIELD_NAMES = tuple(field.split(":", 1)[0] for field in COSTLY_FIELDS)
RATE_LIMIT_HEADERS = ("retry-after", "x-ratelimit-remaining", "x-ratelimit-reset")


def get_github_client(config, preset_id="personal_access_token_credentials_preset"):
Expand All @@ -15,123 +21,165 @@ def get_github_client(config, preset_id="personal_access_token_credentials_prese
def fetch_issues(query_date, github_client, search_query, records_limit,
enable_auto_retry=False, number_of_fetch_retry=0,
fetch_additional_costly_fields=False,
link_to_users=None, user_handle=None, unique_issues_ids=[], current_attempt=1):
current_number_of_fetched_issues = len(unique_issues_ids)
results = []
new_unique_issues_ids = []
try:
logging.info(
"Fetching Issues corresponding to search query '{}' {}".format(
search_query,
_build_retry_log(current_attempt, enable_auto_retry, number_of_fetch_retry))
)
searched_issues = github_client.search_issues(query=search_query)
for issue in searched_issues:
if records_limit != -1 and 0 <= records_limit <= current_number_of_fetched_issues:
logging.info("Limit of {} reached.".format(records_limit))
break
new_record = _build_base_issue_record(issue, query_date)
issue_already_processed = _handle_user_link(new_record, user_handle, link_to_users, unique_issues_ids, new_unique_issues_ids)
if issue_already_processed:
continue
_handle_costly_fields(fetch_additional_costly_fields, issue, new_record)
results.append(new_record)
current_number_of_fetched_issues += 1
except (GithubException, RateLimitExceededException) as fetch_exception:
if not _is_retryable_fetch_exception(fetch_exception):
_raise_unexpected_exception(fetch_exception)
sleep_or_throw_because_of_retryable_fetch_failure(
enable_auto_retry, number_of_fetch_retry, current_attempt, github_client, fetch_exception
)
return fetch_issues(query_date, github_client, search_query, records_limit,
enable_auto_retry, number_of_fetch_retry,
fetch_additional_costly_fields,
link_to_users, user_handle, unique_issues_ids, current_attempt + 1)
except Exception as err:
_raise_unexpected_exception(err)
unique_issues_ids += new_unique_issues_ids
return results


def sleep_or_throw_because_of_retryable_fetch_failure(enable_auto_retry, number_of_fetch_retry, current_attempt,
github_client, fetch_exception):
logging.error(fetch_exception)
now = datetime.utcnow()
retry_log = _build_retry_log(current_attempt, enable_auto_retry, number_of_fetch_retry)
retry_reason = _get_retry_reason(fetch_exception)
status_code = getattr(fetch_exception, "status", "n/a")
if retry_reason == "rate limit":
search_rate_limit = github_client.get_rate_limit().search
logging.info("Data only partially fetched. Retry reason: {}. Status code: {}. Rate limits: {}. Current time: {} {}".format(
retry_reason, status_code, _to_rate_limit_dict(search_rate_limit), now, retry_log
))
else:
logging.info("Data only partially fetched. Retry reason: {}. Status code: {}. Current time: {} {}".format(
retry_reason, status_code, now, retry_log
))
infinite_retry = enable_auto_retry and number_of_fetch_retry == -1
disable_auto_retry = not enable_auto_retry or number_of_fetch_retry == 0
if disable_auto_retry or (not infinite_retry and current_attempt >= number_of_fetch_retry):
logging.info("Could not fetch result due to {} (status {}) even after {}.".format(
retry_reason, status_code, retry_log
))
raise fetch_exception

if retry_reason == "rate limit":
seconds_before_reset = (search_rate_limit.reset - now).total_seconds() + 5
else:
# A 502/503/504 from GitHub search is a transient upstream failure. Retrying is safe because
# the connector is read-only and the existing retry controls still bound the behavior.
seconds_before_reset = 5
logging.info("Sleeping {} seconds before next attempt to fetch data after {} (status {}).".format(
seconds_before_reset, retry_reason, status_code
))
time.sleep(seconds_before_reset)
link_to_users=None, user_handle=None, unique_issues_ids=[]):
max_attempts = _max_attempts(enable_auto_retry, number_of_fetch_retry)
secondary_failures = 0
for attempt in range(1, max_attempts + 1):
current_number_of_fetched_issues = len(unique_issues_ids)
results = []
new_unique_issues_ids = []
try:
logging.info("Fetching Issues corresponding to search query '%s' %s", search_query,
_build_retry_log(attempt, enable_auto_retry, number_of_fetch_retry))
searched_issues = github_client.search_issues(query=search_query)
for issue in searched_issues:
if records_limit != -1 and 0 <= records_limit <= current_number_of_fetched_issues:
logging.info("Limit of %s reached.", records_limit)
break
new_record = _build_base_issue_record(issue, query_date)
issue_already_processed = _handle_user_link(new_record, user_handle, link_to_users,
unique_issues_ids, new_unique_issues_ids)
if issue_already_processed:
continue
_fetch_costly_fields_with_retry(fetch_additional_costly_fields, issue, new_record,
enable_auto_retry, number_of_fetch_retry)
results.append(new_record)
current_number_of_fetched_issues += 1
except GithubException as fetch_exception:
kind, headers = _classify_failure(fetch_exception)
if kind is None:
logging.exception("Unexpected GitHub search failure; propagating error")
raise
if attempt >= max_attempts:
logging.exception("GitHub search retries disabled or exhausted (attempt %s/%s, %s); propagating error",
attempt, max_attempts, kind)
raise
secondary_failures = secondary_failures + 1 if kind in ("secondary", "ambiguous") else 0
_wait_before_retry(fetch_exception, kind, headers, secondary_failures, attempt, max_attempts, "search")
continue
except Exception:
logging.exception("Unexpected error while fetching issues; propagating error")
raise
unique_issues_ids.extend(new_unique_issues_ids)
return results


def _max_attempts(enable_auto_retry, number_of_fetch_retry):
if not enable_auto_retry or number_of_fetch_retry == 0:
return 1
if number_of_fetch_retry == -1:
return MAX_NUMBER_OF_FETCH_ATTEMPT_TO_BYPASS_RATE_LIMIT
return max(1, number_of_fetch_retry)


def _build_retry_log(attempt_number, retry_boolean, max_retry):
return "(attempt {attempt_number}, auto retry {retry_boolean}, max number of retry {max_retry})".format(
attempt_number=attempt_number, retry_boolean=retry_boolean, max_retry=max_retry)


def _is_retryable_fetch_exception(fetch_exception):
return _is_rate_limit_exception(fetch_exception) or _is_transient_server_failure(fetch_exception)


def _get_retry_reason(fetch_exception):
if _is_rate_limit_exception(fetch_exception):
return "rate limit"
return "transient 5xx"


def _is_rate_limit_exception(fetch_exception):
return isinstance(fetch_exception, RateLimitExceededException) or (
isinstance(fetch_exception, GithubException) and
fetch_exception.status == 403 and
"rate limit" in _get_github_exception_message(fetch_exception)
)
def _fetch_costly_fields_with_retry(fetch_additional_costly_fields, issue, new_record,
enable_auto_retry, number_of_fetch_retry):
if not fetch_additional_costly_fields:
return
new_record.update({name: None for name in COSTLY_FIELD_NAMES})
max_attempts = _max_attempts(enable_auto_retry, number_of_fetch_retry)
secondary_failures = 0
for attempt in range(1, max_attempts + 1):
costly_fields = {}
try:
_handle_costly_fields(issue, costly_fields)
new_record.update(costly_fields)
return
except GithubException as error:
kind, headers = _classify_failure(error)
if kind not in ("primary", "secondary", "ambiguous"):
logging.exception("Optional fields failed with a non-rate-limit GitHub error; keeping base record")
return
if attempt >= max_attempts:
logging.exception("Optional-field rate-limit retries disabled or exhausted (attempt %s/%s); keeping base record",
attempt, max_attempts)
return
secondary_failures = secondary_failures + 1 if kind in ("secondary", "ambiguous") else 0
_wait_before_retry(error, kind, headers, secondary_failures, attempt, max_attempts, "optional fields")
except Exception:
logging.exception("Optional fields failed unexpectedly; keeping base record")
return


def _classify_failure(exc):
raw = getattr(exc, "headers", None)
headers = {str(k).lower(): str(v) for k, v in raw.items() if str(k).lower() in RATE_LIMIT_HEADERS} if hasattr(raw, "items") else {}
data = getattr(exc, "data", None)
message = data.get("message", "") if isinstance(data, dict) else ""
message = message if isinstance(message, str) else ""
normalized_message = message.lower()
status = getattr(exc, "status", None)
limited = (isinstance(exc, RateLimitExceededException) or status == 429 or
(status == 403 and ("rate limit" in normalized_message or
headers.get("x-ratelimit-remaining") == "0" or "retry-after" in headers)))
if limited:
if "secondary" in normalized_message or "abuse" in normalized_message:
kind = "secondary"
else:
kind = "primary" if headers.get("x-ratelimit-remaining") == "0" else "ambiguous"
else:
kind = "transient" if status in (502, 503, 504) else None
logging.info("GitHub failure: status=%s message=%s headers=%s rate_limited=%s classification=%s",
status, message, headers, limited, kind)
return kind, headers


def _is_transient_server_failure(fetch_exception):
return isinstance(fetch_exception, GithubException) and fetch_exception.status in [502, 503, 504]
def _parse_nonnegative_header_number(value):
try:
value = float(value)
return value if math.isfinite(value) and value >= 0 else None
except (TypeError, ValueError, OverflowError):
return None


def _retry_delay(kind, headers, secondary_failures, now):
if kind == "transient":
logging.info("Transient GitHub error: retrying in 5 seconds")
return 5
retry_after = _parse_nonnegative_header_number(headers.get("retry-after"))
if retry_after is None and headers.get("retry-after"):
try:
parsed = parsedate_to_datetime(headers["retry-after"])
if parsed.tzinfo is not None:
retry_after = max(0, parsed.timestamp() - now)
except (TypeError, ValueError, OverflowError, OSError):
pass
reset = _parse_nonnegative_header_number(headers.get("x-ratelimit-reset"))
if retry_after is not None:
minimum, reason = retry_after, "Retry-After"
elif headers.get("x-ratelimit-remaining") == "0" and reset is not None:
minimum, reason = max(0, reset - now) + 5, "rate-limit reset plus five seconds"
else:
minimum, reason = 60, "fallback"
if kind in ("secondary", "ambiguous"):
backoff = 60 * 2 ** min(max(secondary_failures - 1, 0), 5)
delay = max(minimum, backoff)
logging.info("Rate-limit delay: %s=%s seconds, %s backoff=%s seconds; retrying in %s seconds",
reason, minimum, kind, backoff, delay)
return delay
logging.info("Rate-limit delay: %s; retrying in %s seconds", reason, minimum)
return minimum


def _get_github_exception_message(fetch_exception):
if not isinstance(fetch_exception, GithubException):
return ""
data = getattr(fetch_exception, "data", {})
if isinstance(data, dict):
return data.get("message", "").lower()
return ""
def _wait_before_retry(error, kind, headers, secondary_failures, attempt, max_attempts, scope):
delay = _retry_delay(kind, headers, secondary_failures, time.time())
logging.warning("GitHub %s failure: status=%s classification=%s headers=%s attempt=%s/%s; sleeping %s seconds before retry",
scope, getattr(error, "status", None), kind, headers, attempt, max_attempts, delay)
time.sleep(delay)


def _handle_costly_fields(fetch_additional_costly_fields, issue_handle, new_record):
if not fetch_additional_costly_fields:
return
def _handle_costly_fields(issue_handle, new_record):
pull_request = issue_handle.as_pull_request()._rawData
_enrich_with_column_values(pull_request, new_record, ["merged", "requested_reviewers", "requested_teams", "merged_at"])
new_record["comments"] = pull_request["comments"] + pull_request["review_comments"]
_enrich_with_column_values(pull_request, new_record, ["base_branch:base.ref", "head_branch:head.ref"])
for field in COSTLY_FIELDS:
if field == COSTLY_COMMENTS_FIELD:
new_record[COSTLY_COMMENTS_FIELD] = pull_request["comments"] + pull_request["review_comments"]
else:
_enrich_with_column_values(pull_request, new_record, [field])


def _handle_user_link(new_record, user_handle, link_to_user, unique_issues_ids, new_unique_issues_ids):
Expand Down Expand Up @@ -160,14 +208,6 @@ def _build_unique_id(issue_id, user_handle):
return "{} linked to {}".format(issue_id, user_handle)


def _to_rate_limit_dict(rate_limit):
return {
"limit": rate_limit.limit,
"remaining": rate_limit.remaining,
"reset": rate_limit.reset
}


def _build_base_issue_record(raw_issue, query_date):
issue_raw_data = raw_issue._rawData
result = {"query_date": query_date}
Expand All @@ -193,6 +233,3 @@ def _enrich_with_column_values(record_raw_data, record_to_enrich, column_names):
record_to_enrich[column_label] = column_value


def _raise_unexpected_exception(unexpected_exception):
logging.error("An unexpected exception occurred while fetching issues: %s", unexpected_exception)
raise unexpected_exception