1  """
  2  This script updates the `max_major_version` attribute of each package in a YAML dependencies
  3  specification (e.g. requirements/core-requirements.yaml) to the maximum available version on PyPI.
  4  """
  5  
  6  import os
  7  import re
  8  from datetime import datetime, timedelta, timezone
  9  
 10  import requests
 11  import yaml
 12  from packaging.version import InvalidVersion, Version
 13  
 14  PACKAGE_NAMES = ["tracing", "skinny", "core", "gateway"]
 15  RELEASE_CUTOFF_DAYS = 14
 16  PYPI_URL = os.environ.get("PYPI_URL", "https://pypi.org").rstrip("/")
 17  
 18  
 19  def check_pypi_accessibility() -> None:
 20      try:
 21          response = requests.head(PYPI_URL, timeout=5)
 22          response.raise_for_status()
 23      except requests.exceptions.RequestException:
 24          raise SystemExit(
 25              f"Error: Cannot connect to {PYPI_URL}. "
 26              "If it's not accessible, set the PYPI_URL environment variable to a PyPI proxy URL."
 27          )
 28  
 29  
 30  def get_latest_major_version(package_name: str) -> int | None:
 31      url = f"{PYPI_URL}/pypi/{package_name}/json"
 32      response = requests.get(url)
 33      response.raise_for_status()
 34      data = response.json()
 35      cutoff = datetime.now(tz=timezone.utc) - timedelta(days=RELEASE_CUTOFF_DAYS)
 36      versions = []
 37      for version, distributions in data["releases"].items():
 38          if len(distributions) == 0 or any(d.get("yanked", False) for d in distributions):
 39              continue
 40  
 41          upload_times = [
 42              datetime.fromisoformat(ut.replace("Z", "+00:00"))
 43              for dist in distributions
 44              if (ut := dist.get("upload_time_iso_8601"))
 45          ]
 46          release_date = min(upload_times) if upload_times else None
 47          if not release_date or release_date >= cutoff:
 48              continue
 49  
 50          try:
 51              version = Version(version)
 52          except InvalidVersion:
 53              # Ignore invalid versions such as https://pypi.org/project/pytz/2004d
 54              continue
 55  
 56          if version.is_devrelease or version.is_prerelease:
 57              continue
 58  
 59          versions.append(version)
 60  
 61      return max(versions).major if versions else None
 62  
 63  
 64  def update_max_major_version(raw: str, key: str, old_value: int, new_value: int) -> str:
 65      """
 66      Update the max_major_version value for a specific package using regex.
 67      This preserves comments and formatting exactly as they appear in the file.
 68      """
 69      # Use word boundaries to ensure we match the exact number, not a substring
 70      pattern = rf"(^{re.escape(key)}:.*?max_major_version:)\s+\b{old_value}\b"
 71      updated, count = re.subn(
 72          pattern, rf"\1 {new_value}", raw, count=1, flags=re.DOTALL | re.MULTILINE
 73      )
 74      if count == 0:
 75          raise ValueError(
 76              f"Failed to update {key}.max_major_version from {old_value} to {new_value}. "
 77              "The pattern may not match the YAML structure."
 78          )
 79      return updated
 80  
 81  
 82  def main() -> None:
 83      check_pypi_accessibility()
 84      for package_name in PACKAGE_NAMES:
 85          req_file_path = os.path.join("requirements", package_name + "-requirements.yaml")
 86          with open(req_file_path) as f:
 87              requirements_src = f.read()
 88  
 89          requirements = yaml.safe_load(requirements_src)
 90          updated_src = requirements_src
 91          changes_made = False
 92  
 93          for key, req_info in requirements.items():
 94              pip_release = req_info["pip_release"]
 95              max_major_version = req_info["max_major_version"]
 96              if req_info.get("freeze", False):
 97                  continue
 98              latest_major_version = get_latest_major_version(pip_release)
 99              if latest_major_version is None:
100                  print(f"Skipping {key}: no releases older than {RELEASE_CUTOFF_DAYS}d found")
101                  continue
102              if latest_major_version > max_major_version:
103                  updated_src = update_max_major_version(
104                      updated_src, key, max_major_version, latest_major_version
105                  )
106                  print(f"Updated {key}.max_major_version to {latest_major_version}")
107                  changes_made = True
108  
109          if changes_made:
110              with open(req_file_path, "w") as f:
111                  f.write(updated_src)
112  
113  
114  if __name__ == "__main__":
115      main()