chore: migrate to pydantic v2

fyhertz · fyhertz · commit 02758cfcde2d · 2023-12-11T11:46:08.000+01:00
diff --git a/ops2deb.lock.yml b/ops2deb.lock.yml
@@ -52,12 +52,12 @@
 - url: https://github.com/istio/istio/releases/download/1.17.2/istio-1.17.2-linux-amd64.tar.gz
   sha256: 9e301bf131be05434807623a2d45af6f625c7b159db4f291f5eb12f262259af7
   timestamp: 2023-04-25 12:44:43+00:00
-- url: https://github.com/upciti/ops2deb/releases/download/2.4.0/ops2deb_linux_amd64
-  sha256: e611e8f85f899d4b55777b4212c6ce4f14f2dbedb1c11a3dd44ee1b63be47d7f
-  timestamp: 2023-04-25 12:44:43+00:00
 - url: https://storage.googleapis.com/kubernetes-release/release/v1.27.1/bin/linux/amd64/kubectl
   sha256: 7fe3a762d926fb068bae32c399880e946e8caf3d903078bea9b169dcd5c17f6d
   timestamp: 2023-04-25 12:44:43+00:00
 - url: https://storage.googleapis.com/minikube/releases/v1.30.1/minikube-linux-amd64
   sha256: e53d9e8c31f4c5f683182f5323d3527aa0725f713945c6d081cf71aa548ab388
   timestamp: 2023-04-25 12:44:43+00:00
+- url: https://github.com/upciti/ops2deb/releases/download/2.4.1/ops2deb_linux_amd64
+  sha256: 07e7f518941be9d2d207fd4d7f8b11fa9e9b2340cabeaf64032e8d44773f5d9f
+  timestamp: 2023-12-11 10:39:21+00:00
diff --git a/ops2deb.yml b/ops2deb.yml
@@ -124,7 +124,7 @@
     - minikube-linux-amd64:/usr/bin/minikube
 
 - name: ops2deb
-  version: 2.4.0
+  version: 2.4.1
   homepage: https://github.com/upciti/ops2deb
   summary: Debian packaging tool for portable applications
   description: |-
diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -21,13 +21,13 @@ packages = [
 python = ">=3.10,<3.12"
 httpx = ">=0.20.0"
 semver = "3.0.1"
-"ruamel.yaml" = ">=0.17.16"
+"ruamel.yaml" = "0.17.21"
 python-debian = ">=0.1.42"
 PyYAML = "^6"
 typer = ">=0.6.1"
 aiofiles = ">=0.1.9"
 Jinja2 = "^3"
-pydantic = "^1"
+pydantic = "^2"
 pyinstaller = { version = "5.13.0", optional = true }
 unix-ar = "^0.2.1"
 rich = "*"
@@ -50,8 +50,8 @@ safety = "*"
 types-aiofiles = "*"
 types-PyYAML = "*"
 pytest-asyncio = ">=0.18.0"
-fastapi = "*"
-ruff = "^0.0.282"
+fastapi = ">= 0.100"
+ruff = "*"
 
 [tool.taskipy.tasks]
 check = """
diff --git a/src/ops2deb/apt.py b/src/ops2deb/apt.py
@@ -5,16 +5,17 @@
 
 import httpx
 from debian.deb822 import Packages, Release
-from pydantic import BaseModel, Field, HttpUrl, ValidationError
+from pydantic import BaseModel, HttpUrl, TypeAdapter, ValidationError
 
 from ops2deb import logger
 from ops2deb.client import client_factory
 from ops2deb.exceptions import Ops2debAptError
 
 
+@dataclass(frozen=True)
 class DebianRepository(BaseModel):
-    url: HttpUrl
-    distribution: str = Field(..., regex=r"[a-zA-Z0-9]+")
+    url: str
+    distribution: str
 
 
 @dataclass(frozen=True, order=True)
@@ -64,9 +65,10 @@ def _parse_debian_repository_option(debian_repository: str) -> DebianRepository:
             '"{repository_url} {distribution}"'
         )
     try:
-        return DebianRepository(url=url, distribution=distribution)
-    except ValidationError as e:
-        raise Ops2debAptError(str(e))
+        TypeAdapter(HttpUrl).validate_python(url)
+    except ValidationError:
+        raise Ops2debAptError("Invalid repository URL")
+    return DebianRepository(url=url, distribution=distribution)
 
 
 async def _list_repository_packages(
diff --git a/src/ops2deb/cli.py b/src/ops2deb/cli.py
@@ -337,7 +337,7 @@ def delta(
         state_delta = compute_state_delta(packages, resources.blueprints)
         print_state_delta_as_rich_table(state_delta)
         if output_as_json:
-            print(state_delta.json(sort_keys=True, indent=2))
+            print(state_delta.model_dump_json(indent=2))
     except Ops2debError as e:
         error(e, exit_code)
 
diff --git a/src/ops2deb/delta.py b/src/ops2deb/delta.py
@@ -28,7 +28,7 @@ def compute_state_delta(
     new_slugs = blueprint_slugs - common_slugs
     deleted_slugs = package_slugs - common_slugs
 
-    state_delta = StateDelta.construct(added=[], removed=[])
+    state_delta = StateDelta.model_construct(added=[], removed=[])
     for slug in new_slugs:
         name, version, architecture = slug.split("_")
         state_delta.added.append(DebianRepositoryPackage(name, version, architecture))
diff --git a/src/ops2deb/formatter.py b/src/ops2deb/formatter.py
@@ -45,7 +45,9 @@ def format_description(description: str) -> str:
 def format_blueprint(blueprint: dict[str, Any]) -> dict[str, Any]:
     if blueprint_arch := blueprint.pop("arch", None):
         blueprint["architecture"] = blueprint_arch
-    blueprint = json.loads(Blueprint.construct(**blueprint).json(exclude_defaults=True))
+    blueprint = json.loads(
+        Blueprint.model_validate(blueprint).model_dump_json(exclude_defaults=True)
+    )
     if (blueprint_fetch := blueprint.get("fetch", None)) and len(blueprint_fetch) == 1:
         blueprint["fetch"] = blueprint_fetch["url"]
     if blueprint_desc := blueprint.get("description", None):
diff --git a/src/ops2deb/generator.py b/src/ops2deb/generator.py
@@ -43,7 +43,7 @@ def __init__(
 
     def _render_template(self, template_name: str) -> None:
         template = environment.get_template(f"{template_name}")
-        package = self.blueprint.dict(exclude={"fetch", "script"})
+        package = self.blueprint.model_dump(exclude={"fetch", "script"})
         package.update({"version": self.debian_version})
         template.stream(package=package).dump(str(self.debian_directory / template_name))
 
@@ -204,7 +204,9 @@ def generate(
     packages: list[SourcePackage] = []
     for blueprint in blueprints:
         for arch, version in product(blueprint.architectures(), blueprint.versions()):
-            blueprint = blueprint.copy(update={"architecture": arch, "version": version})
+            blueprint = blueprint.model_copy(
+                update={"architecture": arch, "version": version}
+            )
             configuration_file = resources.get_blueprint_configuration_file(blueprint)
             configuration_directory = configuration_file.path.parent
             package = SourcePackage(blueprint, output_directory, configuration_directory)
diff --git a/src/ops2deb/lockfile.py b/src/ops2deb/lockfile.py
@@ -4,7 +4,7 @@
 from typing import Protocol, Sequence
 
 import yaml
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel, RootModel, ValidationError
 
 from ops2deb.exceptions import Ops2debLockFileError
 from ops2deb.utils import PrettyYAMLDumper
@@ -21,8 +21,7 @@ class LockEntry(BaseModel):
     timestamp: datetime
 
 
-class LockFileModel(BaseModel):
-    __root__: list[LockEntry]
+LockFileModel = RootModel[list[LockEntry]]
 
 
 def get_utc_datetime() -> datetime:
@@ -39,7 +38,7 @@ def __init__(self, lockfile_path: Path) -> None:
             if lockfile_path.exists() is True:
                 with lockfile_path.open("r") as reader:
                     raw_lockfile = yaml.load(reader, yaml.SafeLoader)
-                lockfile = LockFileModel.parse_obj(raw_lockfile).__root__
+                lockfile = LockFileModel.model_validate(raw_lockfile).root
                 self._entries.update({entry.url: entry for entry in lockfile})
         except yaml.YAMLError as e:
             raise Ops2debLockFileError(f"Invalid YAML file.\n{e}")
@@ -92,7 +91,7 @@ def save(self) -> None:
                 entry.timestamp = now
 
         # sort lockfile entries by urls
-        entries = [entry.dict() for entry in self._entries.values()]
+        entries = [entry.model_dump() for entry in self._entries.values()]
         sorted_entries = sorted(entries, key=itemgetter("timestamp", "url"))
 
         with self.lock_file_path.open("w") as output:
diff --git a/src/ops2deb/parser.py b/src/ops2deb/parser.py
@@ -6,15 +6,16 @@
 from typing import Any, Literal, OrderedDict, cast
 
 from pydantic import (
-    AnyHttpUrl,
     BaseModel,
+    ConfigDict,
     Field,
+    GetCoreSchemaHandler,
     PrivateAttr,
     ValidationError,
-    root_validator,
-    validator,
+    field_validator,
+    model_validator,
 )
-from pydantic.fields import ModelField
+from pydantic_core import CoreSchema, core_schema
 from ruamel.yaml import YAML, YAMLError  # type: ignore[attr-defined]
 
 from ops2deb.exceptions import Ops2debParserError
@@ -28,10 +29,10 @@
 
 
 class Base(BaseModel):
-    class Config:
-        extra = "forbid"
-        frozen = True
-        anystr_strip_whitespace = True
+    model_config = ConfigDict(
+        extra="forbid",
+        str_strip_whitespace=True,
+    )
 
 
 class ArchitectureMap(Base):
@@ -41,7 +42,7 @@ class ArchitectureMap(Base):
 
 
 class MultiArchitectureFetch(Base):
-    url: AnyHttpUrl = Field(..., description="URL template of the file")
+    url: str = Field(..., description="URL template of the file")
     targets: ArchitectureMap | None = Field(
         None,
         description="Architecture to target name map. The URL can be templated with "
@@ -54,19 +55,20 @@ class SourceDestinationStr(str):
     destination: str
 
     @classmethod
-    def __get_validators__(cls) -> Any:
-        yield cls.validate
-
-    @classmethod
-    def validate(cls, v: str) -> "SourceDestinationStr":
-        if not isinstance(v, str):
-            raise TypeError("string required")
-        if len(paths := v.split(":")) != 2:
-            raise TypeError("string must have one ':' separator")
-        string = cls(v)
-        string.source = paths[0]
-        string.destination = paths[1]
-        return string
+    def __get_pydantic_core_schema__(
+        cls, source_type: Any, handler: GetCoreSchemaHandler
+    ) -> CoreSchema:
+        def validate(v: str) -> "SourceDestinationStr":
+            if not isinstance(v, str):
+                raise ValueError("string required")
+            if len(paths := v.split(":")) != 2:
+                raise ValueError("string must have one ':' separator")
+            string = cls(v)
+            string.source = paths[0]
+            string.destination = paths[1]
+            return string
+
+        return core_schema.no_info_plain_validator_function(validate)
 
     def __repr__(self) -> str:
         return (
@@ -97,11 +99,11 @@ class Blueprint(Base):
     )
     version: str = Field("", description="Package name")
     revision: str = Field(
-        "1", regex=r"^[1-9][a-z0-9+~]*$", description="Package revision"
+        "1", pattern=r"^[1-9][a-z0-9+~]*$", description="Package revision"
     )
     epoch: int = Field(0, description="Package epoch", ge=0)
     architecture: Architecture = Field("amd64", description="Package architecture")
-    homepage: AnyHttpUrl | None = Field(None, description="Upstream project homepage")
+    homepage: str | None = Field(None, description="Upstream project homepage")
     summary: str = Field(..., description="Package short description, one line only")
     description: str = Field("", description="Package description")
     build_depends: list[str] = Field(
@@ -124,7 +126,7 @@ class Blueprint(Base):
         description="Conflicting packages, for more information read "
         "https://www.debian.org/doc/debian-policy/ch-relationships.html",
     )
-    fetch: AnyHttpUrl | MultiArchitectureFetch | None = Field(
+    fetch: str | MultiArchitectureFetch | None = Field(
         None,
         description="Url of a file (or a file per architecture) to download before "
         "running build and install instructions",
@@ -135,28 +137,22 @@ class Blueprint(Base):
     _uid: int = PrivateAttr()
     _index_in_configuration: int = PrivateAttr()
 
-    @root_validator(pre=False)
-    def _version_must_be_set(cls, values: Any) -> Any:
-        matrix = values.get("matrix", None)
-        if (not matrix or not matrix.versions) and not values.get("version"):
+    @model_validator(mode="after")
+    def _version_and_architecture_cant_be_used_with_matrix(self) -> "Blueprint":
+        matrix = self.matrix
+        for field in "version", "architecture":
+            if field in self.model_fields_set and matrix is not None:
+                if f"{field}s" in matrix.model_fields_set:
+                    raise ValueError(f"'{field}s' cannot be used with '{field}'")
+        if (not matrix or not matrix.versions) and "version" not in self.model_fields_set:
             raise ValueError("Version field is required when versions matrix is not used")
-        if matrix and matrix.versions:
-            values["version"] = matrix.versions[-1]
-        return values
-
-    @validator("architecture", "version", pre=True, always=False)
-    def _check_arch_and_version(cls, v: Any, values: Any, field: ModelField) -> Any:
-        if (matrix := values.get("matrix", None)) is not None:
-            getter = getattr if isinstance(matrix, Matrix) else lambda x, y: x[y]
-            if getter(matrix, f"{field.name}s"):
-                raise ValueError(f"'{field.name}s' cannot be used with '{field.name}'")
-        return v
-
-    @validator("name", "version", "summary", "description", "homepage", pre=True)
-    def _render_string_attributes(cls, v: Any) -> Any:
-        if isinstance(v, str):
-            return environment.from_string(v).render()
-        return v
+        return self
+
+    @field_validator("revision", mode="before")
+    def _convert_revision_to_str(cls, value: Any) -> Any:
+        if isinstance(value, int):
+            return str(value)
+        return value
 
     def _get_additional_variables(self, architecture: str) -> dict[str, str | None]:
         target = (
@@ -170,6 +166,17 @@ def _get_additional_variables(self, architecture: str) -> dict[str, str | None]:
             rust_target=DEFAULT_RUST_TARGET_MAP.get(architecture, None),
         )
 
+    @classmethod
+    def build(cls, obj: Any) -> "Blueprint":
+        blueprint = cls.model_validate(obj)
+        if blueprint.matrix and (versions := blueprint.matrix.versions):
+            blueprint.version = versions[-1]
+        for string in "name", "version", "summary", "description", "homepage":
+            value = getattr(blueprint, string, None)
+            if isinstance(value, str):
+                setattr(blueprint, string, environment.from_string(value).render())
+        return blueprint
+
     def architectures(self) -> list[str]:
         if self.matrix and self.matrix.architectures:
             return cast(list[str], self.matrix.architectures)
@@ -262,7 +269,7 @@ def load_configuration_file(configuration_path: Path) -> ConfigurationFile:
     blueprints: list[Blueprint] = []
     for index, raw_blueprint in enumerate(raw_blueprints):
         try:
-            blueprint = Blueprint.parse_obj(raw_blueprint)
+            blueprint = Blueprint.build(raw_blueprint)
             blueprint._index_in_configuration = index
             blueprints.append(blueprint)
         except ValidationError as e:
@@ -351,7 +358,7 @@ def load_resources(configurations_search_pattern: str) -> Resources:
             lock = LockFile(lockfile_path)
         locks[lockfile_path] = lock
 
-    # parse blueprints from all configuration files
+    # list all blueprints, and their metadata (associated lockfile and config file)
     blueprints: list[Blueprint] = []
     metadatas: list[BlueprintMetadata] = []
     for configuration_file in configuration_files:
diff --git a/src/ops2deb/utils.py b/src/ops2deb/utils.py
@@ -35,7 +35,7 @@ def expect_block_sequence(self) -> None:
     def choose_scalar_style(self) -> str:
         style: str = super().choose_scalar_style()
         style = '"' if style == "'" else style
-        style = "|" if self.analysis.multiline else style
+        style = "|" if self.analysis and self.analysis.multiline else style
         return style
 
 
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -158,11 +158,11 @@ def blueprint_factory():
     )
 
     def _blueprint_factory(construct: bool = False, **kwargs):
-        blueprint_dict = blueprint.dict(exclude_defaults=True) | kwargs
+        blueprint_dict = blueprint.model_dump(exclude_defaults=True) | kwargs
         if construct is False:
             return Blueprint(**blueprint_dict)
         else:
-            return Blueprint.construct(**blueprint_dict)
+            return Blueprint.model_construct(**blueprint_dict)
 
     return _blueprint_factory
 
diff --git a/tests/test_apt.py b/tests/test_apt.py
@@ -14,8 +14,8 @@ def test_list_repository_packages__should_raise_exception_when_http_error_occurs
 def test_list_repository_packages__should_raise_exception_when_repo_url_is_invalid(
     mock_httpx_client,
 ):
-    with pytest.raises(Ops2debAptError, match="invalid or missing URL scheme"):
-        list_repository_packages("invalid-url strable")
+    with pytest.raises(Ops2debAptError, match="Invalid repository URL"):
+        list_repository_packages("invalid-url stable")
 
 
 def test_list_repository_packages__should_raise_exception_when_repo_distribution_is_missing(  # noqa: E501
diff --git a/tests/test_formatter.py b/tests/test_formatter.py
diff --git a/tests/test_generator.py b/tests/test_generator.py
diff --git a/tests/test_ops2deb.py b/tests/test_ops2deb.py
diff --git a/tests/test_parser.py b/tests/test_parser.py
