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.9.2
 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.9.2,<2.13",
   "requests>=2.24.0,<2.33",
   "tomli>=2.0.0,<2.1;python_version<'3.11'",
 ]

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

@@ -1,5 +1,7 @@
 import typing as t
 
+from pulp_glue.common import oas
+
 
 class AuthProviderBase:
     """
@@ -18,26 +20,32 @@ def can_complete_mutualTLS(self) -> bool:
     def can_complete_oauth2_client_credentials(self, scopes: list[str]) -> bool:
         return False
 
-    def can_complete_scheme(self, scheme: dict[str, t.Any], scopes: list[str]) -> bool:
-        if scheme["type"] == "http":
-            if scheme["scheme"] == "basic":
+    def can_complete_scheme(self, security_scheme: oas.SecurityScheme, scopes: list[str]) -> bool:
+        if isinstance(security_scheme, oas.SecuritySchemeHttp):
+            if security_scheme.scheme == "basic":
                 return self.can_complete_http_basic()
-        elif scheme["type"] == "mutualTLS":
+        elif isinstance(security_scheme, oas.SecuritySchemeMutualTLS):
             return self.can_complete_mutualTLS()
-        elif scheme["type"] == "oauth2":
-            for flow_name, flow in scheme["flows"].items():
-                if flow_name == "clientCredentials" and self.can_complete_oauth2_client_credentials(
-                    flow["scopes"]
-                ):
-                    return True
+        elif isinstance(security_scheme, oas.SecuritySchemeOAuth2):
+            client_credentials_flow = security_scheme.flows.client_credentials
+            if client_credentials_flow is not None and self.can_complete_oauth2_client_credentials(
+                list(client_credentials_flow.scopes.keys())
+            ):
+                return True
         return False
 
     def can_complete(
-        self, proposal: dict[str, list[str]], security_schemes: dict[str, dict[str, t.Any]]
+        self,
+        proposal: dict[str, list[str]],
+        security_schemes: dict[str, oas.SecurityScheme | oas.Reference],
     ) -> bool:
         for name, scopes in proposal.items():
-            scheme = security_schemes.get(name)
-            if scheme is None or not self.can_complete_scheme(scheme, scopes):
+            security_scheme = security_schemes.get(name)
+            if (
+                security_scheme is None
+                or isinstance(security_scheme, oas.Reference)
+                or not self.can_complete_scheme(security_scheme, scopes)
+            ):
                 return False
         # This covers the case where `[]` allows for no auth at all.
         return True

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`."""

…lue/src/pulp_glue/common/pydantic_oas.py pulp-glue/src/pulp_glue/common/oas.pypulp-glue/src/pulp_glue/common/pydantic_oas.py renamed to pulp-glue/src/pulp_glue/common/oas.py pulp-glue/src/pulp_glue/common/pydantic_oas.py renamed to pulp-glue/src/pulp_glue/common/oas.py

@@ -1,4 +1,5 @@
 import typing as t
+from functools import cached_property
 
 import pydantic
 import pydantic.alias_generators
@@ -15,7 +16,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 +31,31 @@ def _check_extensions(self) -> t.Self:
         return self
 
 
+OperationName = t.Literal[
+    "get",
+    "put",
+    "post",
+    "patch",
+    "delete",
+    "options",
+    "head",
+    "trace",
+]
+
+METHODS: list[OperationName] = [
+    "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
 
@@ -164,11 +188,48 @@ class XML(ExtensibleOASBase):
     wrapped: bool = False
 
 
-class Schema(OASBase, extra="allow"):
+class SchemaBase(OASBase, extra="allow"):
     discriminator: Discriminator | None = None
     xml: XML | None = None
     external_docs: ExternalDocumentation | None = None
     example: t.Any = None
+    examples: dict[str, Example | Reference] | None = None
+    nullable: bool = False
+
+
+class AllOfSchema(SchemaBase):
+    all_of: list["Schema"]
+
+
+class AnyOfSchema(SchemaBase):
+    any_of: list["Schema"]
+
+
+class OneOfSchema(SchemaBase):
+    one_of: list["Schema"]
+
+
+class TypeSchema(SchemaBase):
+    type_: str | list[str]
+
+    min_items: int | None = None
+    max_items: int | None = None
+    unique_items: bool = False
+    items: "Schema | None" = None
+
+    minimum: int | float | None = None
+    exclusive_minimum: bool = False
+    maximum: int | float | None = None
+    exclusive_maximum: bool = False
+    multiple_of: int | None = None
+    format_: str | None = None
+    enum: list[str] | None = None
+    properties: dict[str, "Schema"] | None = None
+    additional_properties: "Schema" = True
+    required: list[str] | None = None
+
+
+Schema = bool | Reference | TypeSchema | AllOfSchema | AnyOfSchema | OneOfSchema | SchemaBase
 
 
 class Link(ExtensibleOASBase):
@@ -192,18 +253,38 @@ 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
-    explode: t.Annotated[bool, pydantic.Field(default_factory=lambda data: data["style"] == "form")]
+    explode: t.Annotated[
+        bool,
+        pydantic.Field(
+            # default_factory=lambda data: data["style"] == "form"
+        ),
+    ]
     allow_reserved: bool = False
 
+    @pydantic.model_validator(mode="before")
+    @classmethod
+    def explode_default(cls, data: t.Any) -> t.Any:
+        # This is a workaround for pydantic < 2.10 .
+        # default_factory didn't have the extra data argument.
+        data["explode"] = data.get("style") == "form"
+        return data
+
 
 class MediaType(ExtensibleOASBase):
     schema_: Schema
     example: t.Any = None
     examples: dict[str, Example | Reference] | None = None
+    # TODO encoding seems to be reserved to request body.
     encoding: dict[str, Encoding] = {}
 
 
@@ -237,7 +318,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,17 +332,42 @@ 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"
         ),
     ]
-    explode: t.Annotated[bool, pydantic.Field(default_factory=lambda data: data["style"] == "form")]
     allowed_reserved: bool = False
     example: t.Any = None
     examples: dict[str, Example | Reference] | None = None
 
+    @pydantic.model_validator(mode="before")
+    @classmethod
+    def explode_default(cls, data: t.Any) -> t.Any:
+        # This is a workaround for pydantic < 2.10 .
+        # default_factory didn't have the extra data argument.
+        if "schema" in data:
+            if data.get("style") is None:
+                data["style"] = "simple" if data.get("in") in ["header", "path"] else "form"
+            if data.get("explode") is None:
+                data["explode"] = data["style"] == "form"
+        return data
+
 
 Parameter = ContentParameter | SchemaParameter
 
@@ -275,7 +381,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 +397,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 +417,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] = {}
     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 +440,19 @@ 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
+
+    # @pydantic.computed_field  # type: ignore[prop-decorator]
+    @cached_property
+    def operations(self) -> dict[str, tuple[OperationName, str]]:
+        return {
+            operation.operation_id: (method, path)
+            for path, path_item in self.paths.items()
+            for method, operation in ((method, getattr(path_item, method)) for method in METHODS)
+            if operation is not None
+        }