Webhooks support

Author: p1c2u

openapi_core/security/providers.py

@@ -3,37 +3,37 @@
 
 from openapi_core.security.exceptions import SecurityError
 from openapi_core.spec import Spec
-from openapi_core.validation.request.protocols import Request
+from openapi_core.validation.request.protocols import RequestParameters
 
 
 class BaseProvider:
     def __init__(self, scheme: Spec):
         self.scheme = scheme
 
-    def __call__(self, request: Request) -> Any:
+    def __call__(self, parameters: RequestParameters) -> Any:
         raise NotImplementedError
 
 
 class UnsupportedProvider(BaseProvider):
-    def __call__(self, request: Request) -> Any:
+    def __call__(self, parameters: RequestParameters) -> Any:
         warnings.warn("Unsupported scheme type")
 
 
 class ApiKeyProvider(BaseProvider):
-    def __call__(self, request: Request) -> Any:
+    def __call__(self, parameters: RequestParameters) -> Any:
         name = self.scheme["name"]
         location = self.scheme["in"]
-        source = getattr(request.parameters, location)
+        source = getattr(parameters, location)
         if name not in source:
             raise SecurityError("Missing api key parameter.")
         return source[name]
 
 
 class HttpProvider(BaseProvider):
-    def __call__(self, request: Request) -> Any:
-        if "Authorization" not in request.parameters.header:
+    def __call__(self, parameters: RequestParameters) -> Any:
+        if "Authorization" not in parameters.header:
             raise SecurityError("Missing authorization header.")
-        auth_header = request.parameters.header["Authorization"]
+        auth_header = parameters.header["Authorization"]
         try:
             auth_type, encoded_credentials = auth_header.split(" ", 1)
         except ValueError:

openapi_core/templating/paths/datatypes.py

@@ -2,10 +2,10 @@
 from collections import namedtuple
 
 Path = namedtuple("Path", ["path", "path_result"])
-OperationPath = namedtuple(
-    "OperationPath", ["path", "operation", "path_result"]
+PathOperation = namedtuple(
+    "PathOperation", ["path", "operation", "path_result"]
 )
-ServerOperationPath = namedtuple(
-    "ServerOperationPath",
+PathOperationServer = namedtuple(
+    "PathOperationServer",
     ["path", "operation", "server", "path_result", "server_result"],
 )

openapi_core/templating/paths/finders.py

@@ -10,9 +10,9 @@
 from openapi_core.schema.servers import is_absolute
 from openapi_core.spec import Spec
 from openapi_core.templating.datatypes import TemplateResult
-from openapi_core.templating.paths.datatypes import OperationPath
 from openapi_core.templating.paths.datatypes import Path
-from openapi_core.templating.paths.datatypes import ServerOperationPath
+from openapi_core.templating.paths.datatypes import PathOperation
+from openapi_core.templating.paths.datatypes import PathOperationServer
 from openapi_core.templating.paths.exceptions import OperationNotFound
 from openapi_core.templating.paths.exceptions import PathNotFound
 from openapi_core.templating.paths.exceptions import ServerNotFound
@@ -21,79 +21,87 @@
 from openapi_core.templating.util import search
 
 
-class PathFinder:
+class BasePathFinder:
     def __init__(self, spec: Spec, base_url: Optional[str] = None):
         self.spec = spec
         self.base_url = base_url
 
-    def find(
-        self,
-        method: str,
-        full_url: str,
-    ) -> ServerOperationPath:
-        paths_iter = self._get_paths_iter(full_url)
+    def find(self, method: str, name: str) -> PathOperationServer:
+        paths_iter = self._get_paths_iter(name)
         paths_iter_peek = peekable(paths_iter)
 
         if not paths_iter_peek:
-            raise PathNotFound(full_url)
+            raise PathNotFound(name)
 
-        operations_iter = self._get_operations_iter(paths_iter_peek, method)
+        operations_iter = self._get_operations_iter(method, paths_iter_peek)
         operations_iter_peek = peekable(operations_iter)
 
         if not operations_iter_peek:
-            raise OperationNotFound(full_url, method)
+            raise OperationNotFound(name, method)
 
         servers_iter = self._get_servers_iter(
+            name,
             operations_iter_peek,
-            full_url,
         )
 
         try:
             return next(servers_iter)
         except StopIteration:
-            raise ServerNotFound(full_url)
+            raise ServerNotFound(name)
 
-    def _get_paths_iter(self, full_url: str) -> Iterator[Path]:
+    def _get_paths_iter(self, name: str) -> Iterator[Path]:
+        raise NotImplementedError
+
+    def _get_operations_iter(
+        self, method: str, paths_iter: Iterator[Path]
+    ) -> Iterator[PathOperation]:
+        for path, path_result in paths_iter:
+            if method not in path:
+                continue
+            operation = path / method
+            yield PathOperation(path, operation, path_result)
+
+    def _get_servers_iter(
+        self, name: str, operations_iter: Iterator[PathOperation]
+    ) -> Iterator[PathOperationServer]:
+        raise NotImplementedError
+
+
+class APICallPathFinder(BasePathFinder):
+    def __init__(self, spec: Spec, base_url: Optional[str] = None):
+        self.spec = spec
+        self.base_url = base_url
+
+    def _get_paths_iter(self, name: str) -> Iterator[Path]:
         template_paths: List[Path] = []
         paths = self.spec / "paths"
         for path_pattern, path in list(paths.items()):
             # simple path.
             # Return right away since it is always the most concrete
-            if full_url.endswith(path_pattern):
+            if name.endswith(path_pattern):
                 path_result = TemplateResult(path_pattern, {})
                 yield Path(path, path_result)
             # template path
             else:
-                result = search(path_pattern, full_url)
+                result = search(path_pattern, name)
                 if result:
                     path_result = TemplateResult(path_pattern, result.named)
                     template_paths.append(Path(path, path_result))
 
         # Fewer variables -> more concrete path
         yield from sorted(template_paths, key=template_path_len)
 
-    def _get_operations_iter(
-        self, paths_iter: Iterator[Path], request_method: str
-    ) -> Iterator[OperationPath]:
-        for path, path_result in paths_iter:
-            if request_method not in path:
-                continue
-            operation = path / request_method
-            yield OperationPath(path, operation, path_result)
-
     def _get_servers_iter(
-        self, operations_iter: Iterator[OperationPath], full_url: str
-    ) -> Iterator[ServerOperationPath]:
+        self, name: str, operations_iter: Iterator[PathOperation]
+    ) -> Iterator[PathOperationServer]:
         for path, operation, path_result in operations_iter:
             servers = (
                 path.get("servers", None)
                 or operation.get("servers", None)
                 or self.spec.get("servers", [{"url": "/"}])
             )
             for server in servers:
-                server_url_pattern = full_url.rsplit(path_result.resolved, 1)[
-                    0
-                ]
+                server_url_pattern = name.rsplit(path_result.resolved, 1)[0]
                 server_url = server["url"]
                 if not is_absolute(server_url):
                     # relative to absolute url
@@ -107,7 +115,7 @@ def _get_servers_iter(
                 # simple path
                 if server_url_pattern == server_url:
                     server_result = TemplateResult(server["url"], {})
-                    yield ServerOperationPath(
+                    yield PathOperationServer(
                         path,
                         operation,
                         server,
@@ -121,10 +129,31 @@ def _get_servers_iter(
                         server_result = TemplateResult(
                             server["url"], result.named
                         )
-                        yield ServerOperationPath(
+                        yield PathOperationServer(
                             path,
                             operation,
                             server,
                             path_result,
                             server_result,
                         )
+
+
+class WebhookPathFinder(BasePathFinder):
+    def _get_paths_iter(self, name: str) -> Iterator[Path]:
+        webhooks = self.spec / "webhooks"
+        for webhook_name, path in list(webhooks.items()):
+            if name == webhook_name:
+                path_result = TemplateResult(webhook_name, {})
+                yield Path(path, path_result)
+
+    def _get_servers_iter(
+        self, name: str, operations_iter: Iterator[PathOperation]
+    ) -> Iterator[PathOperationServer]:
+        for path, operation, path_result in operations_iter:
+            yield PathOperationServer(
+                path,
+                operation,
+                None,
+                path_result,
+                {},
+            )

openapi_core/validation/request/__init__.py

@@ -33,6 +33,18 @@
     V31RequestSecurityValidator,
 )
 from openapi_core.validation.request.validators import V31RequestValidator
+from openapi_core.validation.request.validators import (
+    V31WebhookRequestBodyValidator,
+)
+from openapi_core.validation.request.validators import (
+    V31WebhookRequestParametersValidator,
+)
+from openapi_core.validation.request.validators import (
+    V31WebhookRequestSecurityValidator,
+)
+from openapi_core.validation.request.validators import (
+    V31WebhookRequestValidator,
+)
 
 __all__ = [
     "V30RequestBodyValidator",
@@ -43,7 +55,12 @@
     "V31RequestParametersValidator",
     "V31RequestSecurityValidator",
     "V31RequestValidator",
+    "V31WebhookRequestBodyValidator",
+    "V31WebhookRequestParametersValidator",
+    "V31WebhookRequestSecurityValidator",
+    "V31WebhookRequestValidator",
     "V3RequestValidator",
+    "V3WebhookRequestValidator",
     "openapi_v30_request_body_validator",
     "openapi_v30_request_parameters_validator",
     "openapi_v30_request_security_validator",
@@ -64,6 +81,7 @@
 
 # alias to the latest v3 version
 V3RequestValidator = V31RequestValidator
+V3WebhookRequestValidator = V31WebhookRequestValidator
 
 # spec validators
 openapi_v30_request_body_validator = SpecRequestValidatorProxy(

openapi_core/validation/request/exceptions.py

@@ -30,17 +30,11 @@ class MissingRequestBodyError(OpenAPIRequestBodyError):
     """Missing request body error"""
 
 
-@dataclass
 class MissingRequestBody(MissingRequestBodyError):
-    request: Request
-
     def __str__(self) -> str:
         return "Missing request body"
 
 
-@dataclass
 class MissingRequiredRequestBody(MissingRequestBodyError):
-    request: Request
-
     def __str__(self) -> str:
         return "Missing required request body"

openapi_core/validation/request/protocols.py

@@ -15,7 +15,25 @@
 
 
 @runtime_checkable
-class Request(Protocol):
+class BaseRequest(Protocol):
+
+    parameters: RequestParameters
+
+    @property
+    def method(self) -> str:
+        ...
+
+    @property
+    def body(self) -> Optional[str]:
+        ...
+
+    @property
+    def mimetype(self) -> str:
+        ...
+
+
+@runtime_checkable
+class Request(BaseRequest, Protocol):
     """Request attributes protocol.
 
     Attributes:
@@ -44,8 +62,6 @@ class Request(Protocol):
             the mimetype would be "text/html".
     """
 
-    parameters: RequestParameters
-
     @property
     def host_url(self) -> str:
         ...
@@ -54,16 +70,30 @@ def host_url(self) -> str:
     def path(self) -> str:
         ...
 
-    @property
-    def method(self) -> str:
-        ...
 
-    @property
-    def body(self) -> Optional[str]:
-        ...
+@runtime_checkable
+class WebhookRequest(BaseRequest, Protocol):
+    """Webhook request attributes protocol.
+
+    Attributes:
+        name
+            Webhook name
+        method
+            The request method, as lowercase string.
+        parameters
+            A RequestParameters object. Needs to supports path attribute setter
+            to write resolved path parameters.
+        body
+            The request body, as string.
+        mimetype
+            Like content type, but without parameters (eg, without charset,
+            type etc.) and always lowercase.
+            For example if the content type is "text/HTML; charset=utf-8"
+            the mimetype would be "text/html".
+    """
 
     @property
-    def mimetype(self) -> str:
+    def name(self) -> str:
         ...
 
 
@@ -95,3 +125,15 @@ def validate(
         request: Request,
     ) -> RequestValidationResult:
         ...
+
+
+@runtime_checkable
+class WebhookRequestValidator(Protocol):
+    def __init__(self, spec: Spec, base_url: Optional[str] = None):
+        ...
+
+    def validate(
+        self,
+        request: WebhookRequest,
+    ) -> RequestValidationResult:
+        ...