Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
8 changes: 4 additions & 4 deletions pex/private/pex_process_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import venv
from pathlib import Path
from types import SimpleNamespace
from typing import Callable, Optional, Sequence
from typing import Callable, List, Optional, Sequence, Union

RLocationPath = str

Expand Down Expand Up @@ -207,7 +207,7 @@ def post_setup(self, context: SimpleNamespace) -> None:

def create_venv(
venv_name: str,
venv_dir: Path | str,
venv_dir: Union[Path, str],
pth: Sequence[str],
) -> Path:
"""Construct a new Python venv at the requested location.
Expand Down Expand Up @@ -258,7 +258,7 @@ def copy(src: Path, dest: Path) -> None:


def install_files(
imports: list[str], runfiles_manifest: Path, output_dir: Path
imports: List[str], runfiles_manifest: Path, output_dir: Path
) -> None:
"""A helper for installing files in a directory.

Expand Down Expand Up @@ -310,7 +310,7 @@ def main() -> None: # pylint: disable=too-many-locals,too-many-statements
# The new venv is only a couple of files and directories, cleaning
# it up should be fast so it's written to a temp directory.
with tempfile.TemporaryDirectory(
prefix="bzl-pex-", ignore_cleanup_errors=True
prefix="bzl-pex-",
) as tmp:
temp_dir = Path(tmp)

Expand Down
16 changes: 8 additions & 8 deletions pex/private/scie_cache_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
import tempfile
from datetime import datetime
from pathlib import Path
from typing import NamedTuple
from typing import Dict, List, NamedTuple, Optional, Tuple, Union
from urllib.parse import urlparse

# TODO: These values are hard-coded in pex and need to match the current version
Expand Down Expand Up @@ -89,7 +89,7 @@ def verify_download(
target: str,
output_dir: Path,
cache_dir: Path,
extra_args: list[str] | None = None,
extra_args: Optional[List[str]] = None,
) -> None:
"""Run science download and verify it succeeds."""
# Build command with --cache-dir before the download subcommand
Expand Down Expand Up @@ -164,8 +164,8 @@ def populate_hash_cache(
cache_base_dir: Path,
url: str,
source_file: Path,
digest_json: dict[str, str | int] | None = None,
ttl: bool | None = None,
digest_json: Optional[Dict[str, Union[str, int]]] = None,
ttl: Optional[bool] = None,
) -> None:
"""Populate science's hash-based cache directory with a file.

Expand Down Expand Up @@ -240,7 +240,7 @@ def get_binary_version(binary_path: Path) -> str:
return result.stdout.strip()


def parse_interpreter_filename(interpreter_name: str) -> tuple[str, str, str, str]:
def parse_interpreter_filename(interpreter_name: str) -> Tuple[str, str, str, str]:
"""Parse Python interpreter archive filename.

Args:
Expand Down Expand Up @@ -436,7 +436,7 @@ def cache_sha256sums(

def get_interpreter_info(
interpreter: Path,
expected_version: str | None = None,
expected_version: Optional[str] = None,
) -> InterpreterInfo:
"""Extract interpreter information and compute hash.

Expand Down Expand Up @@ -472,7 +472,7 @@ def get_interpreter_info(
)


def _build_distributions_json(interpreter_info: InterpreterInfo) -> dict[str, object]:
def _build_distributions_json(interpreter_info: InterpreterInfo) -> Dict[str, object]:
"""Build the distributions JSON structure.

Args:
Expand Down Expand Up @@ -503,7 +503,7 @@ def _build_distributions_json(interpreter_info: InterpreterInfo) -> dict[str, ob
def cache_distributions_json(
cache_base_dir: Path,
major_minor_version: str,
dist_json: dict[str, object],
dist_json: Dict[str, object],
) -> None:
"""Cache the distributions JSON file.

Expand Down
24 changes: 12 additions & 12 deletions pex/private/update_pex_versions.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
import time
import urllib.request
from pathlib import Path
from typing import Any
from typing import Any, Dict, List, Optional
from urllib.error import HTTPError
from urllib.parse import urlparse
from urllib.request import urlopen
Expand Down Expand Up @@ -99,7 +99,7 @@ def parse_args() -> argparse.Namespace:
return parser.parse_args()


def fetch_sha256(sha256_url: str, artifact_name: str) -> str | None:
def fetch_sha256(sha256_url: str, artifact_name: str) -> Optional[str]:
"""Parse sha256 hash from a .sha256 file for a specific artifact."""
req = urllib.request.Request(sha256_url, headers=REQUEST_HEADERS)
logging.debug("Fetching sha256 file: %s", sha256_url)
Expand Down Expand Up @@ -138,11 +138,11 @@ def integrity(hex_str: str) -> str:


def _process_release_artifacts(
release: dict[str, Any],
release: Dict[str, Any],
version: str,
platforms: list[str],
platforms: List[str],
artifact_prefix: str,
) -> dict[str, dict[str, str]]:
) -> Dict[str, Dict[str, str]]:
"""Process artifacts for a single release."""
assets_map = {asset["name"]: asset for asset in release["assets"]}
artifacts = {}
Expand Down Expand Up @@ -184,10 +184,10 @@ def _process_release_artifacts(


def _process_pex_release_artifacts(
release: dict[str, Any],
release: Dict[str, Any],
version: str,
platforms: list[str],
) -> dict[str, dict[str, str]]:
platforms: List[str],
) -> Dict[str, Dict[str, str]]:
"""Process artifacts for a single pex release (platform-specific .pex files)."""
assets_map = {asset["name"]: asset for asset in release["assets"]}
artifacts = {}
Expand Down Expand Up @@ -246,10 +246,10 @@ def _handle_rate_limit(exc: HTTPError) -> None:


def query_releases(
platforms: list[str],
platforms: List[str],
repo: str,
artifact_prefix: str,
) -> dict[str, dict[str, dict[str, str]]]:
) -> Dict[str, Dict[str, Dict[str, str]]]:
"""Query GitHub releases and extract scie binaries for each platform."""
page = 1
releases_data = {}
Expand Down Expand Up @@ -299,8 +299,8 @@ def query_releases(


def query_pex_releases(
repo: str, platforms: list[str]
) -> dict[str, dict[str, dict[str, str]]]:
repo: str, platforms: List[str]
) -> Dict[str, Dict[str, Dict[str, str]]]:
"""Query GitHub releases and extract pex platform-specific binaries."""
page = 1
releases_data = {}
Expand Down
14 changes: 7 additions & 7 deletions pex/private/update_python_versions.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import tempfile
from enum import IntEnum
from pathlib import Path
from typing import Any, NamedTuple
from typing import Any, Dict, NamedTuple, Optional, Tuple

from python.runfiles import Runfiles

Expand Down Expand Up @@ -124,7 +124,7 @@ def integrity(hex_str: str) -> str:
return f"sha256-{encoded}"


def _platform_from_asset_name(asset_name: str) -> PlatformInfo | None:
def _platform_from_asset_name(asset_name: str) -> Optional[PlatformInfo]:
"""Extract platform identifier from Python Build Standalone asset name.

Examples:
Expand Down Expand Up @@ -172,7 +172,7 @@ def _process_version(
provider: str,
python_version: str,
tmp_path: Path,
) -> dict[str, dict[str, str]] | None:
) -> Optional[Dict[str, Dict[str, str]]]:
"""Process a single Python version for a provider.

Args:
Expand Down Expand Up @@ -201,7 +201,7 @@ def _process_version(

# Process each distribution asset, preferring gnu builds for Linux
# Track builds with preference: gnueabihf > gnu > gnueabi > musl
platform_assets: dict[str, tuple[dict[str, str], LibcPreference]] = {}
platform_assets: Dict[str, Tuple[Dict[str, str], LibcPreference]] = {}

assets = distributions.get("assets", [])
if isinstance(assets, list):
Expand Down Expand Up @@ -235,8 +235,8 @@ def _process_version(


def _process_asset(
asset: dict[str, object], base_url: str
) -> tuple[str, dict[str, str], LibcPreference] | None:
asset: Dict[str, object], base_url: str
) -> Optional[Tuple[str, Dict[str, str], LibcPreference]]:
"""Process a single distribution asset.

Args:
Expand Down Expand Up @@ -288,7 +288,7 @@ def _process_asset(

def download_provider_info(
science_binary: Path, provider: str, version: str, tmp_dir: Path
) -> dict[str, Any] | None:
) -> Optional[Dict[str, Any]]:
"""Download provider information using science download command.

Args:
Expand Down
4 changes: 3 additions & 1 deletion tests/deps/dep_a/dep_a.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
"""A dependency module for testing."""

from typing import List

def get_messages() -> list[str]:

def get_messages() -> List[str]:
"""Return a list with one string."""
return ["Message from dep_a"]
4 changes: 3 additions & 1 deletion tests/deps/dep_b/dep_b.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
"""A dependency module for testing."""

from typing import List

from tests.deps.dep_c.dep_c import get_messages as get_messages_c


def get_messages() -> list[str]:
def get_messages() -> List[str]:
"""Return a list, always adds a special string and extends the same function from dep_c."""
messages = ["Special message from dep_b"]
messages.extend(get_messages_c())
Expand Down
4 changes: 3 additions & 1 deletion tests/deps/dep_c/dep_c.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
"""A dependency module for testing."""

from typing import List

def get_messages() -> list[str]:

def get_messages() -> List[str]:
"""Return a list with one string."""
return ["Message from dep_c"]
Loading