Skip to content
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

style: use type annotations from the future (for apt.py) #149

Merged
merged 1 commit into from
Jan 29, 2025
Merged

style: use type annotations from the future (for apt.py) #149

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
106 changes: 54 additions & 52 deletions lib/charms/operator_libs_linux/v0/apt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,8 @@
```
"""

from __future__ import annotations

import fileinput
import glob
import logging
Expand All @@ -108,7 +110,7 @@
import subprocess
from enum import Enum
from subprocess import PIPE, CalledProcessError, check_output
from typing import Any, Dict, Iterable, Iterator, List, Mapping, Optional, Tuple, Union
from typing import Any, Iterable, Iterator, Mapping
from urllib.parse import urlparse

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -230,8 +232,8 @@ def __str__(self):
@staticmethod
def _apt(
command: str,
package_names: Union[str, List],
optargs: Optional[List[str]] = None,
package_names: str | list,
optargs: list[str] | None = None,
) -> None:
"""Wrap package management commands for Debian/Ubuntu systems.

Expand Down Expand Up @@ -321,7 +323,7 @@ def state(self, state: PackageState) -> None:
self._state = state

@property
def version(self) -> "Version":
def version(self) -> Version:
"""Returns the version for a package."""
return self._version

Expand All @@ -341,16 +343,16 @@ def fullversion(self) -> str:
return "{}.{}".format(self._version, self._arch)

@staticmethod
def _get_epoch_from_version(version: str) -> Tuple[str, str]:
def _get_epoch_from_version(version: str) -> tuple[str, str]:
"""Pull the epoch, if any, out of a version string."""
epoch_matcher = re.compile(r"^((?P<epoch>\d+):)?(?P<version>.*)")
matches = epoch_matcher.search(version).groupdict()
return matches.get("epoch", ""), matches.get("version")

@classmethod
def from_system(
cls, package: str, version: Optional[str] = "", arch: Optional[str] = ""
) -> "DebianPackage":
cls, package: str, version: str | None = "", arch: str | None = ""
) -> DebianPackage:
"""Locates a package, either on the system or known to apt, and serializes the information.

Args:
Expand Down Expand Up @@ -382,8 +384,8 @@ def from_system(

@classmethod
def from_installed_package(
cls, package: str, version: Optional[str] = "", arch: Optional[str] = ""
) -> "DebianPackage":
cls, package: str, version: str | None = "", arch: str | None = ""
) -> DebianPackage:
"""Check whether the package is already installed and return an instance.

Args:
Expand Down Expand Up @@ -452,8 +454,8 @@ def from_installed_package(

@classmethod
def from_apt_cache(
cls, package: str, version: Optional[str] = "", arch: Optional[str] = ""
) -> "DebianPackage":
cls, package: str, version: str | None = "", arch: str | None = ""
) -> DebianPackage:
"""Check whether the package is already installed and return an instance.

Args:
Expand Down Expand Up @@ -542,7 +544,7 @@ def number(self) -> str:
"""Returns the version number for a package."""
return self._version

def _get_parts(self, version: str) -> Tuple[str, str]:
def _get_parts(self, version: str) -> tuple[str, str]:
"""Separate the version into component upstream and Debian pieces."""
try:
version.rindex("-")
Expand All @@ -553,7 +555,7 @@ def _get_parts(self, version: str) -> Tuple[str, str]:
upstream, debian = version.rsplit("-", 1)
return upstream, debian

def _listify(self, revision: str) -> List[str]:
def _listify(self, revision: str) -> list[str]:
"""Split a revision string into a listself.

This list is comprised of alternating between strings and numbers,
Expand All @@ -569,7 +571,7 @@ def _listify(self, revision: str) -> List[str]:
revision = remains
return result

def _get_alphas(self, revision: str) -> Tuple[str, str]:
def _get_alphas(self, revision: str) -> tuple[str, str]:
"""Return a tuple of the first non-digit characters of a revision."""
# get the index of the first digit
for i, char in enumerate(revision):
Expand All @@ -580,7 +582,7 @@ def _get_alphas(self, revision: str) -> Tuple[str, str]:
# string is entirely alphas
return revision, ""

def _get_digits(self, revision: str) -> Tuple[int, str]:
def _get_digits(self, revision: str) -> tuple[int, str]:
"""Return a tuple of the first integer characters of a revision."""
# If the string is empty, return (0,'')
if not revision:
Expand Down Expand Up @@ -722,11 +724,11 @@ def __ne__(self, other) -> bool:


def add_package(
package_names: Union[str, List[str]],
version: Optional[str] = "",
arch: Optional[str] = "",
update_cache: Optional[bool] = False,
) -> Union[DebianPackage, List[DebianPackage]]:
package_names: str | list[str],
version: str | None = "",
arch: str | None = "",
update_cache: bool | None = False,
) -> DebianPackage | list[DebianPackage]:
"""Add a package or list of packages to the system.

Args:
Expand Down Expand Up @@ -784,9 +786,9 @@ def add_package(

def _add(
name: str,
version: Optional[str] = "",
arch: Optional[str] = "",
) -> Tuple[Union[DebianPackage, str], bool]:
version: str | None = "",
arch: str | None = "",
) -> tuple[DebianPackage | str, bool]:
"""Add a package to the system.

Args:
Expand All @@ -806,8 +808,8 @@ def _add(


def remove_package(
package_names: Union[str, List[str]],
) -> Union[DebianPackage, List[DebianPackage]]:
package_names: str | list[str],
) -> DebianPackage | list[DebianPackage]:
"""Remove package(s) from the system.

Args:
Expand All @@ -816,7 +818,7 @@ def remove_package(
Raises:
TypeError: if no packages are provided
"""
packages: List[DebianPackage] = []
packages: list[DebianPackage] = []

package_names = [package_names] if isinstance(package_names, str) else package_names
if not package_names:
Expand Down Expand Up @@ -923,7 +925,7 @@ class GPGKeyError(Error):
class DebianRepository:
"""An abstraction to represent a repository."""

_deb822_stanza: Optional["_Deb822Stanza"] = None
_deb822_stanza: _Deb822Stanza | None = None
"""set by Deb822Stanza after creating a DebianRepository"""

def __init__(
Expand All @@ -932,10 +934,10 @@ def __init__(
repotype: str,
uri: str,
release: str,
groups: List[str],
groups: list[str],
filename: str = "",
gpg_key_filename: str = "",
options: Optional[Dict[str, str]] = None,
options: dict[str, str] | None = None,
):
self._enabled = enabled
self._repotype = repotype
Expand Down Expand Up @@ -1023,7 +1025,7 @@ def prefix_from_uri(uri: str) -> str:
return "/etc/apt/sources.list.d/{}".format(path)

@staticmethod
def from_repo_line(repo_line: str, write_file: Optional[bool] = True) -> "DebianRepository":
def from_repo_line(repo_line: str, write_file: bool | None = True) -> DebianRepository:
"""Instantiate a new `DebianRepository` from a `sources.list` entry line.

Args:
Expand Down Expand Up @@ -1233,10 +1235,10 @@ class RepositoryMapping(Mapping[str, DebianRepository]):
_sources_subdir = "sources.list.d"
_default_list_name = "sources.list"
_default_sources_name = "ubuntu.sources"
_last_errors: Tuple[Error, ...] = ()
_last_errors: tuple[Error, ...] = ()

def __init__(self):
self._repository_map: Dict[str, DebianRepository] = {}
self._repository_map: dict[str, DebianRepository] = {}
self.default_file = os.path.join(self._apt_dir, self._default_list_name)
# ^ public attribute for backwards compatibility only
sources_dir = os.path.join(self._apt_dir, self._sources_subdir)
Expand Down Expand Up @@ -1320,16 +1322,16 @@ def _parse_deb822_lines(
cls,
lines: Iterable[str],
filename: str = "",
) -> Tuple[List[DebianRepository], List[InvalidSourceError]]:
) -> tuple[list[DebianRepository], list[InvalidSourceError]]:
"""Parse lines from a deb822 file into a list of repos and a list of errors.

The semantics of `_parse_deb822_lines` slightly different to `_parse`:
`_parse` reads a commented out line as an entry that is not enabled
`_parse_deb822_lines` strips out comments entirely when parsing a file into stanzas,
instead only reading the 'Enabled' key to determine if an entry is enabled
"""
repos: List[DebianRepository] = []
errors: List[InvalidSourceError] = []
repos: list[DebianRepository] = []
errors: list[InvalidSourceError] = []
for numbered_lines in _iter_deb822_stanzas(lines):
try:
stanza = _Deb822Stanza(numbered_lines=numbered_lines, filename=filename)
Expand All @@ -1345,8 +1347,8 @@ def load(self, filename: str):
Args:
filename: the path to the repository file
"""
parsed: List[int] = []
skipped: List[int] = []
parsed: list[int] = []
skipped: list[int] = []
with open(filename, "r") as f:
for n, line in enumerate(f, start=1): # 1 indexed line numbers
try:
Expand Down Expand Up @@ -1424,7 +1426,7 @@ def _parse(line: str, filename: str) -> DebianRepository:
raise InvalidSourceError("An invalid sources line was found in %s!", filename)

def add( # noqa: D417 # undocumented-param: default_filename intentionally undocumented
self, repo: DebianRepository, default_filename: Optional[bool] = False
self, repo: DebianRepository, default_filename: bool | None = False
) -> None:
"""Add a new repository to the system using add-apt-repository.

Expand Down Expand Up @@ -1513,7 +1515,7 @@ class _Deb822Stanza:
May define multiple DebianRepository objects.
"""

def __init__(self, numbered_lines: List[Tuple[int, str]], filename: str = ""):
def __init__(self, numbered_lines: list[tuple[int, str]], filename: str = ""):
self._filename = filename
self._numbered_lines = numbered_lines
if not numbered_lines:
Expand All @@ -1531,7 +1533,7 @@ def __init__(self, numbered_lines: List[Tuple[int, str]], filename: str = ""):
self._gpg_key_filename, self._gpg_key_from_stanza = gpg_key_info

@property
def repos(self) -> Tuple[DebianRepository, ...]:
def repos(self) -> tuple[DebianRepository, ...]:
"""The repositories defined by this deb822 stanza."""
return self._repos

Expand All @@ -1554,7 +1556,7 @@ def get_gpg_key_filename(self) -> str:
class MissingRequiredKeyError(InvalidSourceError):
"""Missing a required value in a source file."""

def __init__(self, message: str = "", *, file: str, line: Optional[int], key: str) -> None:
def __init__(self, message: str = "", *, file: str, line: int | None, key: str) -> None:
super().__init__(message, file, line, key)
self.file = file
self.line = line
Expand All @@ -1569,7 +1571,7 @@ def __init__(
message: str = "",
*,
file: str,
line: Optional[int],
line: int | None,
key: str,
value: str,
) -> None:
Expand All @@ -1580,7 +1582,7 @@ def __init__(
self.value = value


def _iter_deb822_stanzas(lines: Iterable[str]) -> Iterator[List[Tuple[int, str]]]:
def _iter_deb822_stanzas(lines: Iterable[str]) -> Iterator[list[tuple[int, str]]]:
"""Given lines from a deb822 format file, yield a stanza of lines.

Args:
Expand All @@ -1590,7 +1592,7 @@ def _iter_deb822_stanzas(lines: Iterable[str]) -> Iterator[List[Tuple[int, str]]
lists of numbered lines (a tuple of line number and line) that make up
a deb822 stanza, with comments stripped out (but accounted for in line numbering)
"""
current_stanza: List[Tuple[int, str]] = []
current_stanza: list[tuple[int, str]] = []
for n, line in enumerate(lines, start=1): # 1 indexed line numbers
if not line.strip(): # blank lines separate stanzas
if current_stanza:
Expand All @@ -1605,8 +1607,8 @@ def _iter_deb822_stanzas(lines: Iterable[str]) -> Iterator[List[Tuple[int, str]]


def _deb822_stanza_to_options(
lines: Iterable[Tuple[int, str]],
) -> Tuple[Dict[str, str], Dict[str, int]]:
lines: Iterable[tuple[int, str]],
) -> tuple[dict[str, str], dict[str, int]]:
"""Turn numbered lines into a dict of options and a dict of line numbers.

Args:
Expand All @@ -1616,8 +1618,8 @@ def _deb822_stanza_to_options(
a dictionary of option names to (potentially multiline) values, and
a dictionary of option names to starting line number
"""
parts: Dict[str, List[str]] = {}
line_numbers: Dict[str, int] = {}
parts: dict[str, list[str]] = {}
line_numbers: dict[str, int] = {}
current = None
for n, line in lines:
assert "#" not in line # comments should be stripped out
Expand All @@ -1634,8 +1636,8 @@ def _deb822_stanza_to_options(


def _deb822_options_to_repos(
options: Dict[str, str], line_numbers: Mapping[str, int] = {}, filename: str = ""
) -> Tuple[Tuple[DebianRepository, ...], Tuple[str, Optional[str]]]:
options: dict[str, str], line_numbers: Mapping[str, int] = {}, filename: str = ""
) -> tuple[tuple[DebianRepository, ...], tuple[str, str | None]]:
"""Return a collections of DebianRepository objects defined by this deb822 stanza.

Args:
Expand Down Expand Up @@ -1666,7 +1668,7 @@ def _deb822_options_to_repos(
)
# Signed-By
gpg_key_file = options.pop("Signed-By", "")
gpg_key_from_stanza: Optional[str] = None
gpg_key_from_stanza: str | None = None
if "\n" in gpg_key_file:
# actually a literal multi-line gpg-key rather than a filename
gpg_key_from_stanza = gpg_key_file
Expand All @@ -1687,7 +1689,7 @@ def _deb822_options_to_repos(
# suite can specify an exact path, in which case the components must be omitted and suite must end with a slash (/).
# If suite does not specify an exact path, at least one component must be present.
# https://manpages.ubuntu.com/manpages/noble/man5/sources.list.5.html
components: List[str]
components: list[str]
if len(suites) == 1 and suites[0].endswith("/"):
if "Components" in options:
msg = (
Expand Down