Implement --upload-before

Author: uranusjr

pyproject.toml

@@ -259,7 +259,7 @@ max-complexity = 33  # default is 10
 [tool.ruff.lint.pylint]
 max-args = 15  # default is 5
 max-branches = 28  # default is 12
-max-returns = 13  # default is 6
+max-returns = 14  # default is 6
 max-statements = 134  # default is 50
 
 ######################################################################################

src/pip/_internal/cli/cmdoptions.py

@@ -11,6 +11,7 @@
 # mypy: strict-optional=False
 from __future__ import annotations
 
+import datetime
 import importlib.util
 import logging
 import os
@@ -796,6 +797,34 @@ def _handle_dependency_group(
     help="Ignore the Requires-Python information.",
 )
 
+
+def _handle_upload_before(
+    option: Option, opt: str, value: str, parser: OptionParser
+) -> None:
+    """
+    Process a value provided for the --upload-before option.
+
+    This is an optparse.Option callback for the --upload-before option.
+    """
+    if value is None:
+        return None
+    upload_before = datetime.datetime.fromisoformat(value)
+    # Assume local timezone if no offset is given in the ISO string.
+    if upload_before.tzinfo is None:
+        upload_before = upload_before.astimezone()
+    parser.values.upload_before = upload_before
+
+
+upload_before: Callable[..., Option] = partial(
+    Option,
+    "--upload-before",
+    dest="upload_before",
+    metavar="datetime",
+    action="callback",
+    callback=_handle_upload_before,
+    help="Skip uploads after given time. This should be an ISO 8601 string.",
+)
+
 no_build_isolation: Callable[..., Option] = partial(
     Option,
     "--no-build-isolation",

src/pip/_internal/cli/req_command.py

@@ -5,8 +5,6 @@
 PackageFinder machinery and all its vendored dependencies, etc.
 """
 
-from __future__ import annotations
-
 import logging
 from functools import partial
 from optparse import Values
@@ -328,6 +326,7 @@ def _build_package_finder(
         session: PipSession,
         target_python: TargetPython | None = None,
         ignore_requires_python: bool | None = None,
+        upload_before: datetime.datetime | None = None,
     ) -> PackageFinder:
         """
         Create a package finder appropriate to this requirement command.
@@ -348,4 +347,5 @@ def _build_package_finder(
             link_collector=link_collector,
             selection_prefs=selection_prefs,
             target_python=target_python,
+            upload_before=upload_before,
         )

src/pip/_internal/commands/download.py

@@ -51,6 +51,7 @@ def add_options(self) -> None:
         self.cmd_opts.add_option(cmdoptions.no_use_pep517())
         self.cmd_opts.add_option(cmdoptions.check_build_deps())
         self.cmd_opts.add_option(cmdoptions.ignore_requires_python())
+        self.cmd_opts.add_option(cmdoptions.upload_before())
 
         self.cmd_opts.add_option(
             "-d",
@@ -93,6 +94,7 @@ def run(self, options: Values, args: list[str]) -> int:
             session=session,
             target_python=target_python,
             ignore_requires_python=options.ignore_requires_python,
+            upload_before=options.upload_before,
         )
 
         build_tracker = self.enter_context(get_build_tracker())

src/pip/_internal/commands/index.py

@@ -1,6 +1,4 @@
-from __future__ import annotations
-
-import json
+import datetime
 import logging
 from collections.abc import Iterable
 from optparse import Values
@@ -40,6 +38,7 @@ def add_options(self) -> None:
         cmdoptions.add_target_python_options(self.cmd_opts)
 
         self.cmd_opts.add_option(cmdoptions.ignore_requires_python())
+        self.cmd_opts.add_option(cmdoptions.upload_before())
         self.cmd_opts.add_option(cmdoptions.pre())
         self.cmd_opts.add_option(cmdoptions.json())
         self.cmd_opts.add_option(cmdoptions.no_binary())
@@ -86,6 +85,7 @@ def _build_package_finder(
         session: PipSession,
         target_python: TargetPython | None = None,
         ignore_requires_python: bool | None = None,
+        upload_before: datetime.datetime | None = None,
     ) -> PackageFinder:
         """
         Create a package finder appropriate to the index command.
@@ -103,6 +103,7 @@ def _build_package_finder(
             link_collector=link_collector,
             selection_prefs=selection_prefs,
             target_python=target_python,
+            upload_before=upload_before,
         )
 
     def get_available_package_versions(self, options: Values, args: list[Any]) -> None:
@@ -118,6 +119,7 @@ def get_available_package_versions(self, options: Values, args: list[Any]) -> No
                 session=session,
                 target_python=target_python,
                 ignore_requires_python=options.ignore_requires_python,
+                upload_before=options.upload_before,
             )
 
             versions: Iterable[Version] = (

src/pip/_internal/commands/install.py

@@ -207,6 +207,7 @@ def add_options(self) -> None:
             ),
         )
 
+        self.cmd_opts.add_option(cmdoptions.upload_before())
         self.cmd_opts.add_option(cmdoptions.ignore_requires_python())
         self.cmd_opts.add_option(cmdoptions.no_build_isolation())
         self.cmd_opts.add_option(cmdoptions.use_pep517())
@@ -344,6 +345,7 @@ def run(self, options: Values, args: list[str]) -> int:
             session=session,
             target_python=target_python,
             ignore_requires_python=options.ignore_requires_python,
+            upload_before=options.upload_before,
         )
         build_tracker = self.enter_context(get_build_tracker())
 

src/pip/_internal/commands/list.py

@@ -143,8 +143,10 @@ def handle_pip_version_check(self, options: Values) -> None:
             super().handle_pip_version_check(options)
 
     def _build_package_finder(
-        self, options: Values, session: PipSession
-    ) -> PackageFinder:
+        self,
+        options: Values,
+        session: "PipSession",
+    ) -> "PackageFinder":
         """
         Create a package finder appropriate to this list command.
         """

src/pip/_internal/commands/wheel.py

@@ -64,6 +64,7 @@ def add_options(self) -> None:
         self.cmd_opts.add_option(cmdoptions.requirements())
         self.cmd_opts.add_option(cmdoptions.src())
         self.cmd_opts.add_option(cmdoptions.ignore_requires_python())
+        self.cmd_opts.add_option(cmdoptions.upload_before())
         self.cmd_opts.add_option(cmdoptions.no_deps())
         self.cmd_opts.add_option(cmdoptions.progress_bar())
 
@@ -103,7 +104,11 @@ def add_options(self) -> None:
     def run(self, options: Values, args: list[str]) -> int:
         session = self.get_default_session(options)
 
-        finder = self._build_package_finder(options, session)
+        finder = self._build_package_finder(
+            options=options,
+            session=session,
+            upload_before=options.upload_before,
+        )
 
         options.wheel_dir = normalize_path(options.wheel_dir)
         ensure_dir(options.wheel_dir)

src/pip/_internal/index/package_finder.py

@@ -1,7 +1,5 @@
 """Routines related to PyPI, indexes"""
 
-from __future__ import annotations
-
 import enum
 import functools
 import itertools
@@ -111,6 +109,7 @@ class LinkType(enum.Enum):
     format_invalid = enum.auto()
     platform_mismatch = enum.auto()
     requires_python_mismatch = enum.auto()
+    upload_too_late = enum.auto()
 
 
 class LinkEvaluator:
@@ -131,7 +130,8 @@ def __init__(
         formats: frozenset[str],
         target_python: TargetPython,
         allow_yanked: bool,
-        ignore_requires_python: bool | None = None,
+        ignore_requires_python: Optional[bool] = None,
+        upload_before: Optional[datetime.datetime] = None,
     ) -> None:
         """
         :param project_name: The user supplied package name.
@@ -149,6 +149,7 @@ def __init__(
         :param ignore_requires_python: Whether to ignore incompatible
             PEP 503 "data-requires-python" values in HTML links. Defaults
             to False.
+        :param upload_before: If set, only allow links prior to the given date.
         """
         if ignore_requires_python is None:
             ignore_requires_python = False
@@ -158,6 +159,7 @@ def __init__(
         self._ignore_requires_python = ignore_requires_python
         self._formats = formats
         self._target_python = target_python
+        self._upload_before = upload_before
 
         self.project_name = project_name
 
@@ -176,6 +178,11 @@ def evaluate_link(self, link: Link) -> tuple[LinkType, str]:
             reason = link.yanked_reason or "<none given>"
             return (LinkType.yanked, f"yanked for reason: {reason}")
 
+        if link.upload_time is not None and self._upload_before is not None:
+            if link.upload_time > self._upload_before:
+                reason = f"Upload time {link.upload_time} after {self._upload_before}"
+                return (LinkType.upload_too_late, reason)
+
         if link.egg_fragment:
             egg_info = link.egg_fragment
             ext = link.ext
@@ -590,9 +597,10 @@ def __init__(
         link_collector: LinkCollector,
         target_python: TargetPython,
         allow_yanked: bool,
-        format_control: FormatControl | None = None,
-        candidate_prefs: CandidatePreferences | None = None,
-        ignore_requires_python: bool | None = None,
+        format_control: Optional[FormatControl] = None,
+        candidate_prefs: Optional[CandidatePreferences] = None,
+        ignore_requires_python: Optional[bool] = None,
+        upload_before: Optional[datetime.datetime] = None,
     ) -> None:
         """
         This constructor is primarily meant to be used by the create() class
@@ -614,6 +622,7 @@ def __init__(
         self._ignore_requires_python = ignore_requires_python
         self._link_collector = link_collector
         self._target_python = target_python
+        self._upload_before = upload_before
 
         self.format_control = format_control
 
@@ -636,15 +645,17 @@ def create(
         cls,
         link_collector: LinkCollector,
         selection_prefs: SelectionPreferences,
-        target_python: TargetPython | None = None,
-    ) -> PackageFinder:
+        target_python: Optional[TargetPython] = None,
+        upload_before: Optional[datetime.datetime] = None,
+    ) -> "PackageFinder":
         """Create a PackageFinder.
 
         :param selection_prefs: The candidate selection preferences, as a
             SelectionPreferences object.
         :param target_python: The target Python interpreter to use when
             checking compatibility. If None (the default), a TargetPython
             object will be constructed from the running Python.
+        :param upload_before: If set, only find links prior to the given date.
         """
         if target_python is None:
             target_python = TargetPython()
@@ -661,6 +672,7 @@ def create(
             allow_yanked=selection_prefs.allow_yanked,
             format_control=selection_prefs.format_control,
             ignore_requires_python=selection_prefs.ignore_requires_python,
+            upload_before=upload_before,
         )
 
     @property
@@ -739,6 +751,7 @@ def make_link_evaluator(self, project_name: str) -> LinkEvaluator:
             target_python=self._target_python,
             allow_yanked=self._allow_yanked,
             ignore_requires_python=self._ignore_requires_python,
+            upload_before=self._upload_before,
         )
 
     def _sort_links(self, links: Iterable[Link]) -> list[Link]:

src/pip/_internal/models/link.py

@@ -1,5 +1,4 @@
-from __future__ import annotations
-
+import datetime
 import functools
 import itertools
 import logging
@@ -207,17 +206,19 @@ class Link:
         "requires_python",
         "yanked_reason",
         "metadata_file_data",
+        "upload_time",
         "cache_link_parsing",
         "egg_fragment",
     ]
 
     def __init__(
         self,
         url: str,
-        comes_from: str | IndexContent | None = None,
-        requires_python: str | None = None,
-        yanked_reason: str | None = None,
-        metadata_file_data: MetadataFile | None = None,
+        comes_from: Optional[Union[str, "IndexContent"]] = None,
+        requires_python: Optional[str] = None,
+        yanked_reason: Optional[str] = None,
+        metadata_file_data: Optional[MetadataFile] = None,
+        upload_time: Optional[datetime.datetime] = None,
         cache_link_parsing: bool = True,
         hashes: Mapping[str, str] | None = None,
     ) -> None:
@@ -239,6 +240,8 @@ def __init__(
             no such metadata is provided. This argument, if not None, indicates
             that a separate metadata file exists, and also optionally supplies
             hashes for that file.
+        :param upload_time: upload time of the file, or None if the information
+            is not available from the server.
         :param cache_link_parsing: A flag that is used elsewhere to determine
             whether resources retrieved from this link should be cached. PyPI
             URLs should generally have this set to False, for example.
@@ -272,6 +275,7 @@ def __init__(
         self.requires_python = requires_python if requires_python else None
         self.yanked_reason = yanked_reason
         self.metadata_file_data = metadata_file_data
+        self.upload_time = upload_time
 
         self.cache_link_parsing = cache_link_parsing
         self.egg_fragment = self._egg_fragment()
@@ -300,6 +304,12 @@ def from_json(
         if metadata_info is None:
             metadata_info = file_data.get("dist-info-metadata")
 
+        upload_time: Optional[datetime.datetime]
+        if upload_time_data := file_data.get("upload-time"):
+            upload_time = datetime.datetime.fromisoformat(upload_time_data)
+        else:
+            upload_time = None
+
         # The metadata info value may be a boolean, or a dict of hashes.
         if isinstance(metadata_info, dict):
             # The file exists, and hashes have been supplied
@@ -325,6 +335,7 @@ def from_json(
             yanked_reason=yanked_reason,
             hashes=hashes,
             metadata_file_data=metadata_file_data,
+            upload_time=upload_time,
         )
 
     @classmethod