OpenAPI app and high level integration

openapi_core/app.py

@@ -0,0 +1,65 @@
+from pathlib import Path
+
+from jsonschema_path import SchemaPath
+from jsonschema_path.handlers.protocols import SupportsRead
+from jsonschema_path.typing import Schema
+from openapi_spec_validator.validation.types import SpecValidatorType
+
+from openapi_core.protocols import Integration
+from openapi_core.shortcuts import unmarshal_request
+from openapi_core.shortcuts import unmarshal_response
+from openapi_core.shortcuts import validate_request
+from openapi_core.shortcuts import validate_response
+from openapi_core.typing import RequestType
+from openapi_core.typing import ResponseType
+from openapi_core.unmarshalling.request.datatypes import RequestUnmarshalResult
+from openapi_core.unmarshalling.response.datatypes import ResponseUnmarshalResult
+
+
+class OpenAPI:
+
+    def __init__(self, spec: SchemaPath, integration: Integration, spec_validator_cls: SpecValidatorType):
+        if spec_validator_cls is not None:
+            spec_validator = spec_validator_cls(spec)
+            spec_validator.validate()
+
+        self.spec = spec
+        self.integration = integration
+
+    @classmethod
+    def from_dict(cls, data: Schema, integration: Integration, spec_validator_cls: SpecValidatorType) -> "OpenAPI":
+        spec = SchemaPath.from_dict(data)
+        return cls(spec, integration, spec_validator_cls=spec_validator_cls)
+
+    @classmethod
+    def from_path(cls, path: Path, integration: Integration, spec_validator_cls: SpecValidatorType) -> "OpenAPI":
+        spec = SchemaPath.from_path(path)
+        return cls(spec, integration, spec_validator_cls=spec_validator_cls)
+
+    @classmethod
+    def from_file_path(cls, file_path: str, integration: Integration, spec_validator_cls: SpecValidatorType) -> "OpenAPI":
+        spec = SchemaPath.from_file_path(file_path)
+        return cls(spec, integration, spec_validator_cls=spec_validator_cls)
+
+    @classmethod
+    def from_file(cls, fileobj: SupportsRead, integration: Integration, spec_validator_cls: SpecValidatorType) -> "OpenAPI":
+        spec = SchemaPath.from_file(fileobj)
+        return cls(spec, integration, spec_validator_cls=spec_validator_cls)
+
+    def validate_request(self, request: RequestType) -> None:
+        openapi_request = self.integration.get_openapi_request(request)
+        validate_request(openapi_request, spec=self.spec)
+
+    def validate_response(self, request: RequestType, response: ResponseType) -> None:
+        openapi_request = self.integration.get_openapi_request(request)
+        openapi_response = self.integration.get_openapi_response(response)
+        validate_response(openapi_request, openapi_response, spec=self.spec)
+
+    def unmarshal_request(self, request: RequestType) -> RequestUnmarshalResult:
+        openapi_request = self.integration.get_openapi_request(request)
+        return unmarshal_request(openapi_request, spec=self.spec)
+
+    def unmarshal_response(self, request: RequestType, response: ResponseType) -> ResponseUnmarshalResult:
+        openapi_request = self.integration.get_openapi_request(request)
+        openapi_response = self.integration.get_openapi_response(response)
+        return unmarshal_response(openapi_request, openapi_response, spec=self.spec)

openapi_core/contrib/flask/integrations.py

@@ -0,0 +1,17 @@
+from flask.wrappers import Request
+from flask.wrappers import Response
+
+from openapi_core.contrib.flask.requests import FlaskOpenAPIRequest
+from openapi_core.contrib.flask.responses import FlaskOpenAPIResponse
+
+
+class FlaskIntegration:
+    request_cls = FlaskOpenAPIRequest
+    response_cls = FlaskOpenAPIResponse
+
+    def get_openapi_request(self, request: Request) -> FlaskOpenAPIRequest:
+        return self.request_cls(request)
+
+    def get_openapi_response(self, response: Response) -> FlaskOpenAPIResponse:
+        assert self.response_cls is not None
+        return self.response_cls(response)

openapi_core/protocols.py

@@ -6,6 +6,8 @@
 from typing import runtime_checkable
 
 from openapi_core.datatypes import RequestParameters
+from openapi_core.typing import RequestType
+from openapi_core.typing import ResponseType
 
 
 @runtime_checkable
@@ -134,3 +136,14 @@ def content_type(self) -> str:
     @property
     def headers(self) -> Mapping[str, Any]:
         ...
+
+
+@runtime_checkable
+class Integration(Protocol):
+    """Integration protocol."""
+
+    def get_openapi_request(self, request: RequestType) -> Request:
+        ...
+
+    def get_openapi_response(self, response: ResponseType) -> Response:
+        ...