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
 
+import pulp_glue.common.pydantic_oas as s
+
 
 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: s.SecurityScheme, scopes: list[str]) -> bool:
+        if isinstance(security_scheme, s.SecuritySchemeHttp):
+            if security_scheme.scheme == "basic":
                 return self.can_complete_http_basic()
-        elif scheme["type"] == "mutualTLS":
+        elif isinstance(security_scheme, s.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, s.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, s.SecurityScheme | s.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, s.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`."""