Skip to content

Commit

Permalink
feat: upgrade to pydantic 2.x
Browse files Browse the repository at this point in the history
  • Loading branch information
Anas Husseini committed Jan 11, 2024
1 parent ea04ada commit cc218d6
Show file tree
Hide file tree
Showing 4 changed files with 98 additions and 82 deletions.
2 changes: 1 addition & 1 deletion craft_archives/repo/apt_sources_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,7 @@ def _install_sources_apt(
formats=cast(Optional[List[str]], package_repo.formats),
name=name,
suites=suites,
url=package_repo.url,
url=str(package_repo.url).rstrip("/"),
keyring_path=keyring_path,
)

Expand Down
148 changes: 80 additions & 68 deletions craft_archives/repo/package_repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,21 +22,24 @@
from typing import Any, Dict, List, Mapping, Optional, Union
from urllib.parse import urlparse

import pydantic
from overrides import overrides # pyright: ignore[reportUnknownVariableType]
from pydantic import (
AnyUrl,
ConstrainedStr,
BaseModel,
ConfigDict,
Field,
FileUrl,
root_validator, # pyright: ignore[reportUnknownVariableType]
validator, # pyright: ignore[reportUnknownVariableType]
StringConstraints,
ValidationInfo,
field_validator, # pyright: ignore[reportUnknownVariableType]
model_validator, # pyright: ignore[reportUnknownVariableType]
)

# NOTE: using this instead of typing.Literal because of this bad typing_extensions
# interaction: https://github.com/pydantic/pydantic/issues/5821#issuecomment-1559196859
# We can revisit this when typing_extensions >4.6.0 is released, and/or we no longer
# have to support Python <3.10
from typing_extensions import Literal
# We can revisit this when typing_extensions >4.6.0 is released, and/or we no
# longer have to support Python <3.10
from typing_extensions import Annotated, Literal

from . import errors

Expand All @@ -47,14 +50,6 @@
UCA_KEY_ID = "391A9AA2147192839E9DB0315EDB1B62EC4926EA"


class KeyIdStr(ConstrainedStr):
"""A constrained string for a GPG key ID."""

min_length = 40
max_length = 40
regex = re.compile(r"^[0-9A-F]{40}$")


class PriorityString(enum.IntEnum):
"""Convenience values that represent common deb priorities."""

Expand All @@ -70,22 +65,22 @@ def _alias_generator(value: str) -> str:
return value.replace("_", "-")


class PackageRepository(pydantic.BaseModel, abc.ABC):
class PackageRepository(BaseModel, abc.ABC):
"""The base class for package repositories."""

class Config: # pylint: disable=too-few-public-methods
"""Pydantic model configuration."""

validate_assignment = True
allow_mutation = False
allow_population_by_field_name = True
alias_generator = _alias_generator
extra = "forbid"
model_config = ConfigDict(
validate_assignment=True,
frozen=True,
populate_by_name=True,
alias_generator=_alias_generator,
extra="forbid",
)

type: Literal["apt"]
priority: Optional[PriorityValue]
priority: Optional[PriorityValue] = None

@root_validator
@model_validator(mode="before")
@classmethod
def priority_cannot_be_zero(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""Priority cannot be zero per apt Preferences specification."""
priority = values.get("priority")
Expand All @@ -96,9 +91,10 @@ def priority_cannot_be_zero(cls, values: Dict[str, Any]) -> Dict[str, Any]:
)
return values

@validator("priority")
@field_validator("priority")
@classmethod
def _convert_priority_to_int(
cls, priority: Optional[PriorityValue], values: Dict[str, Any]
cls, priority: Optional[PriorityValue], info: ValidationInfo
) -> Optional[int]:
if isinstance(priority, str):
str_priority = priority.upper()
Expand All @@ -107,7 +103,11 @@ def _convert_priority_to_int(
# This cannot happen; if it's a string but not one of the accepted
# ones Pydantic will fail early and won't call this validator.
raise _create_validation_error(
url=str(values.get("url") or values.get("ppa") or values.get("cloud")),
url=str(
info.data.get("url")
or info.data.get("ppa")
or info.data.get("cloud")
),
message=(
f"invalid priority {priority!r}. "
"Priority must be 'always', 'prefer', 'defer' or a nonzero integer."
Expand All @@ -117,7 +117,7 @@ def _convert_priority_to_int(

def marshal(self) -> Dict[str, Union[str, int]]:
"""Return the package repository data as a dictionary."""
return self.dict(by_alias=True, exclude_none=True)
return self.model_dump(by_alias=True, exclude_none=True)

@classmethod
def unmarshal(cls, data: Mapping[str, Any]) -> "PackageRepository":
Expand Down Expand Up @@ -173,7 +173,8 @@ class PackageRepositoryAptPPA(PackageRepository):

ppa: str

@validator("ppa")
@field_validator("ppa")
@classmethod
def _non_empty_ppa(cls, ppa: str) -> str:
if not ppa:
raise _create_validation_error(
Expand All @@ -200,7 +201,8 @@ class PackageRepositoryAptUCA(PackageRepository):
cloud: str
pocket: Literal["updates", "proposed"] = "updates"

@validator("cloud")
@field_validator("cloud")
@classmethod
def _non_empty_cloud(cls, cloud: str) -> str:
if not cloud:
raise _create_validation_error(message="clouds must be non-empty strings.")
Expand All @@ -222,71 +224,81 @@ class PackageRepositoryApt(PackageRepository):
"""An APT package repository."""

url: Union[AnyUrl, FileUrl]
key_id: KeyIdStr = pydantic.Field(alias="key-id")
architectures: Optional[List[str]]
formats: Optional[List[Literal["deb", "deb-src"]]]
path: Optional[str]
components: Optional[List[str]]
key_server: Optional[str] = pydantic.Field(alias="key-server")
suites: Optional[List[str]]

# Customize some of the validation error messages
class Config(PackageRepository.Config): # noqa: D106 - no docstring needed
error_msg_templates = {
"value_error.any_str.min_length": "Invalid URL; URLs must be non-empty strings"
}
key_id: Annotated[
str, StringConstraints(min_length=40, max_length=40, pattern=r"^[0-9A-F]{40}$")
] = Field(alias="key-id")
architectures: Optional[List[str]] = None
formats: Optional[List[Literal["deb", "deb-src"]]] = None
path: Optional[str] = None
components: Optional[List[str]] = None
key_server: Optional[Annotated[str, Field(alias="key-server")]] = None
suites: Optional[List[str]] = None

@property
def name(self) -> str:
"""Get the repository name."""
return re.sub(r"\W+", "_", self.url)
return re.sub(r"\W+", "_", str(self.url).rstrip("/"))

@validator("path")
@field_validator("url")
@classmethod
def _convert_url_to_string(cls, url: Union[AnyUrl, FileUrl]) -> str:
return str(url).rstrip("/")

@field_validator("path")
@classmethod
def _path_non_empty(
cls, path: Optional[str], values: Dict[str, Any]
cls, path: Optional[str], info: ValidationInfo
) -> Optional[str]:
if path is not None and not path:
raise _create_validation_error(
url=values.get("url"),
url=info.data.get("url"),
message="Invalid path; Paths must be non-empty strings.",
)
return path

@validator("components")
@field_validator("components")
@classmethod
def _not_mixing_components_and_path(
cls, components: Optional[List[str]], values: Dict[str, Any]
cls, components: Optional[List[str]], info: ValidationInfo
) -> Optional[List[str]]:
path = values.get("path")
path = info.data.get("path")
if components and path:
raise _create_validation_error(
url=values.get("url"),
url=info.data.get("url"),
message=(
f"components {components!r} cannot be combined with "
f"path {path!r}."
),
)
return components

@validator("suites")
@field_validator("suites")
@classmethod
def _not_mixing_suites_and_path(
cls, suites: Optional[List[str]], values: Dict[str, Any]
cls, suites: Optional[List[str]], info: ValidationInfo
) -> Optional[List[str]]:
path = values.get("path")
path = info.data.get("path")
if suites and path:
message = f"suites {suites!r} cannot be combined with path {path!r}."
raise _create_validation_error(url=values.get("url"), message=message)
raise _create_validation_error(url=info.data.get("url"), message=message)
return suites

@validator("suites", each_item=True)
def _suites_without_backslash(cls, suite: str, values: Dict[str, Any]) -> str:
if suite.endswith("/"):
raise _create_validation_error(
url=values.get("url"),
message=f"invalid suite {suite!r}. Suites must not end with a '/'.",
)
return suite
@field_validator("suites")
@classmethod
def _suites_without_backslash(
cls, suites: List[str], info: ValidationInfo
) -> List[str]:
for suite in suites:
if suite.endswith("/"):
raise _create_validation_error(
url=info.data.get("url"),
message=f"invalid suite {suite!r}. Suites must not end "
"with a '/'.",
)
return suites

@root_validator
@model_validator(mode="before")
@classmethod
def _missing_components_or_suites(cls, values: Dict[str, Any]) -> Dict[str, Any]:
suites = values.get("suites")
components = values.get("components")
Expand All @@ -311,12 +323,12 @@ def unmarshal(cls, data: Mapping[str, Any]) -> "PackageRepositoryApt":
@property
def pin(self) -> str:
"""The pin string for this repository if needed."""
domain = urlparse(self.url).netloc
domain = urlparse(str(self.url).rstrip("/")).netloc
return f'origin "{domain}"'


def _create_validation_error(*, url: Optional[str] = None, message: str) -> ValueError:
"""Create a ValueError with a formatted message and an optional indicative ``url``."""
"""Create a ValueError with a formatted message and an optional url."""
error_message = ""
if url:
error_message += f"Invalid package repository for '{url}': "
Expand Down
6 changes: 3 additions & 3 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ dependencies = [
"lazr.restfulclient",
"lazr.uri",
"overrides",
"pydantic<2.0.0",
"pydantic>=2.0.0,<3.0.0",
]
classifiers = [
"Development Status :: 3 - Alpha",
Expand Down Expand Up @@ -159,8 +159,8 @@ extend-exclude = [
"__pycache__",
]
pep8-naming.classmethod-decorators = [
"pydantic.validator",
"pydantic.root_validator"
"pydantic.field_validator",
"pydantic.model_validator"
]

# Follow ST063 - Maintaining and updating linting specifications for updating these.
Expand Down
24 changes: 14 additions & 10 deletions tests/unit/repo/test_package_repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ def test_apt_valid_architectures(arch):

def test_apt_invalid_url():
with pytest.raises(
pydantic.ValidationError, match="Invalid URL; URLs must be non-empty strings"
pydantic.ValidationError, match="Input should be a valid URL, input is empty"
):
create_apt(
key_id="A" * 40,
Expand All @@ -156,12 +156,13 @@ def test_apt_invalid_path_with_suites():
create_apt(
key_id="A" * 40,
path="/",
components=["main"],
suites=["xenial", "xenial-updates"],
url="http://archive.ubuntu.com/ubuntu",
)

expected_message = (
"Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"Value error, Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"suites ['xenial', 'xenial-updates'] cannot be combined with path '/'"
)

Expand All @@ -175,11 +176,12 @@ def test_apt_invalid_path_with_components():
key_id="A" * 40,
path="/",
components=["main"],
suites=["xenial", "xenial-updates"],
url="http://archive.ubuntu.com/ubuntu",
)

expected_message = (
"Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"Value error, Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"components ['main'] cannot be combined with path '/'."
)

Expand All @@ -196,7 +198,7 @@ def test_apt_invalid_missing_components():
)

expected_message = (
"Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"Value error, Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"components must be specified when using suites."
)

Expand All @@ -213,7 +215,7 @@ def test_apt_invalid_missing_suites():
)

expected_message = (
"Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"Value error, Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"suites must be specified when using components."
)

Expand All @@ -225,13 +227,14 @@ def test_apt_invalid_suites_as_path():
with pytest.raises(pydantic.ValidationError) as raised:
create_apt(
key_id="A" * 40,
components=["main"],
suites=["my-suite/"],
url="http://archive.ubuntu.com/ubuntu",
)

expected_message = (
"Invalid package repository for 'http://archive.ubuntu.com/ubuntu': "
"invalid suite 'my-suite/'. Suites must not end with a '/'."
"Value error, Invalid package repository for 'http://archive.ubuntu.com/"
"ubuntu': invalid suite 'my-suite/'. Suites must not end with a '/'."
)

err = raised.value
Expand Down Expand Up @@ -260,7 +263,8 @@ def test_apt_key_id_invalid(key_id):
"key-id": key_id,
}

with pytest.raises(pydantic.ValidationError, match="string does not match regex"):
error = r"String should match pattern '\^\[0-9A-F\]\{40\}\$'"
with pytest.raises(pydantic.ValidationError, match=error):
PackageRepositoryApt.unmarshal(repo)


Expand All @@ -285,7 +289,7 @@ def test_apt_formats(formats):
apt_deb = PackageRepositoryApt.unmarshal(repo)
assert apt_deb.formats == formats
else:
error = ".*unexpected value; permitted: 'deb', 'deb-src'"
error = "Input should be 'deb' or 'deb-src'"
with pytest.raises(pydantic.ValidationError, match=error):
PackageRepositoryApt.unmarshal(repo)

Expand Down Expand Up @@ -379,7 +383,7 @@ def test_apt_invalid_priority():
create_apt(key_id="A" * 40, url="http://test", priority=0)

expected_message = (
"Invalid package repository for 'http://test': "
"Value error, Invalid package repository for 'http://test': "
"invalid priority: Priority cannot be zero."
)

Expand Down

0 comments on commit cc218d6

Please sign in to comment.