Use pydantic in openapi spec

Author: mdellweg

lower_bounds_constraints.lock

@@ -7,6 +7,6 @@ schema==0.7.5
 tomli==2.0.0
 tomli-w==1.0.0
 pygments==2.17.2
-pydantic==2.11.7
+pydantic==2.8.1
 click-shell==2.1
 SecretStorage==3.3.3

pulp-glue/pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 dependencies = [
   "multidict>=6.0.5,<6.8",
   "packaging>=22.0,<=26.0",  # CalVer
-  "pydantic>=2.11.7,<2.13",
+  "pydantic>=2.8.1,<2.13",
   "requests>=2.24.0,<2.33",
   "tomli>=2.0.0,<2.1;python_version<'3.11'",
 ]

pulp-glue/src/pulp_glue/common/exceptions.py

@@ -59,5 +59,9 @@ class ValidationError(OpenAPIError):
     """Exception raised for failed client side validation of parameters or request bodies."""
 
 
+class SchemaError(OpenAPIError):
+    """Exception raised for unsurmountable inconsistencies of the openapi schema."""
+
+
 class UnsafeCallError(OpenAPIError):
     """Exception raised for POST, PUT, PATCH or DELETE calls with `safe_calls_only=True`."""

pulp-glue/src/pulp_glue/common/openapi.py

@@ -15,7 +15,7 @@ class OASBase(pydantic.BaseModel, alias_generator=to_alias, extra="forbid"):
 
 class ExtensibleOASBase(OASBase, extra="allow"):
     @pydantic.model_validator(mode="after")
-    def _check_extensions(self) -> t.Self:
+    def _check_extensions(self) -> "t.Self":
         if self.__pydantic_extra__ is not None:
             invalid_keys = [
                 key for key in self.__pydantic_extra__.keys() if not key.startswith("x-")
@@ -30,8 +30,20 @@ def _check_extensions(self) -> t.Self:
         return self
 
 
+OperationName = t.Literal[
+    "get",
+    "put",
+    "post",
+    "patch",
+    "delete",
+    "options",
+    "head",
+    "trace",
+]
+
+
 class Reference(OASBase):
-    _ref: t.Annotated[str, pydantic.Field(alias="$ref")]
+    ref: t.Annotated[str, pydantic.Field(alias="$ref")]
     summary: str | None = None
     description: str | None = None
 
@@ -192,7 +204,13 @@ class Encoding(ExtensibleOASBase):
     # TODO These only apply to RFC6570
     style: (
         t.Literal[
-            "simple", "form", "matrix", "label", "spaceDelimited", "pipeDelimited", "deepObject"
+            "simple",
+            "form",
+            "matrix",
+            "label",
+            "spaceDelimited",
+            "pipeDelimited",
+            "deepObject",
         ]
         | None
     ) = None
@@ -237,7 +255,7 @@ class ParameterBase(ExtensibleOASBase):
     allow_empty_value: bool = False
 
     @pydantic.model_validator(mode="after")
-    def _path_required(self) -> t.Self:
+    def _path_required(self) -> "t.Self":
         if self.in_ == "path":
             assert self.required
         return self
@@ -251,10 +269,18 @@ class SchemaParameter(ParameterBase):
     schema_: Schema
     style: t.Annotated[
         t.Literal[
-            "simple", "form", "matrix", "label", "spaceDelimited", "pipeDelimited", "deepObject"
+            "simple",
+            "form",
+            "matrix",
+            "label",
+            "spaceDelimited",
+            "pipeDelimited",
+            "deepObject",
         ],
         pydantic.Field(
-            default_factory=lambda data: "simple" if data["in_"] in ["header", "path"] else "form"
+            default_factory=lambda data: (
+                "simple" if data.get("in_") in ["header", "path"] else "form"
+            )
         ),
     ]
     explode: t.Annotated[bool, pydantic.Field(default_factory=lambda data: data["style"] == "form")]
@@ -275,7 +301,7 @@ class RequestBody(ExtensibleOASBase):
 class Response(ExtensibleOASBase):
     description: str
     headers: dict[str, Header | Reference] | None = None
-    content: dict[str, MediaType] | None = None
+    content: dict[str, MediaType] = {}
     links: dict[str, Link | Reference] | None = None
 
 
@@ -291,16 +317,17 @@ class Operation(ExtensibleOASBase):
     description: str | None = None
     external_docs: ExternalDocumentation | None = None
     operation_id: str
-    parameters: list[Parameter | Reference] | None = None
+    parameters: list[Parameter | Reference] = []
     request_body: RequestBody | Reference | None = None
-    responses: Responses | None = None
+    responses: Responses = {}
     callbacks: dict[str, Callback | Reference] | None = None
     deprecated: bool = False
     security: SecurityRequirements | None = None
     servers: list[Server] | None = None
 
 
-class PathItem(Reference, ExtensibleOASBase):
+class PathItem(ExtensibleOASBase):
+    # TODO $ref
     get: Operation | None = None
     put: Operation | None = None
     post: Operation | None = None
@@ -310,17 +337,17 @@ class PathItem(Reference, ExtensibleOASBase):
     head: Operation | None = None
     trace: Operation | None = None
     servers: list[Server] | None = None
-    parameters: list[Parameter | Reference] | None = None
+    parameters: list[Parameter | Reference] = []
 
 
 class Components(ExtensibleOASBase):
-    schemas: dict[str, Schema] | None = None
+    schemas: dict[str, Schema] | None = {}
     responses: dict[str, Response | Reference] | None = None
-    parameters: dict[str, Parameter | Reference] | None = None
+    parameters: dict[str, Parameter | Reference] = {}
     examples: dict[str, Example | Reference] | None = None
     request_bodies: dict[str, RequestBody | Reference] | None = None
     headers: dict[str, Header | Reference] | None = None
-    security_schemes: dict[str, SecurityScheme | Reference] | None = None
+    security_schemes: dict[str, SecurityScheme | Reference] = {}
     links: dict[str, Link | Reference] | None = None
     callbacks: dict[str, Callback | Reference] | None = None
     path_items: dict[str, PathItem] | None = None
@@ -333,9 +360,9 @@ class OpenAPISpec(ExtensibleOASBase):
     servers: list[Server] | None = None
     # In the specification there is a Paths Object,
     # probably because paths as keys can get extra validation.
-    paths: dict[str, PathItem] | None = None
+    paths: dict[str, PathItem] = {}
     webhooks: dict[str, PathItem] | None = None
-    components: Components | None = None
+    components: Components = Components()
     security: SecurityRequirements | None = None
     tags: list[Tag] | None = None
     external_docs: ExternalDocumentation | None = None

pulp-glue/src/pulp_glue/common/pydantic_oas.py

@@ -4,6 +4,7 @@
 import typing as t
 from contextlib import suppress
 
+from pulp_glue.common.exceptions import SchemaError, ValidationError
 from pulp_glue.common.i18n import get_translation
 
 translation = get_translation(__package__)
@@ -13,14 +14,6 @@
 ISO_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
 
 
-class SchemaError(ValueError):
-    pass
-
-
-class ValidationError(ValueError):
-    pass
-
-
 class OpenApi3JsonEncoder(json.JSONEncoder):
     def default(self, o: t.Any) -> t.Any:
         if isinstance(o, datetime.datetime):