1import json
  2from functools import cached_property
  3from typing import Any, ClassVar
  4
  5from plain.http import Request, Response
  6from plain.views.exceptions import ResponseException
  7
  8from .views import APIView
  9
 10__all__ = [
 11    "APIVersionChange",
 12    "VersionedAPIView",
 13]
 14
 15
 16class APIVersionChange:
 17    description: str = ""
 18
 19    def transform_request_forward(self, request: Request, data: dict[str, Any]) -> None:
 20        """
 21        If this version of the API made a change in how a request is processed,
 22        (ex. the name of an input changed) then you can
 23        """
 24
 25    def transform_response_backward(
 26        self, response: Response, data: dict[str, Any]
 27    ) -> None:
 28        """
 29        Transform the response data for this version.
 30
 31        We only transform the response data if we are moving backward to an older version.
 32        This is because the response data is always in the latest version.
 33        """
 34
 35
 36class VersionedAPIView(APIView):
 37    # API versions from newest to oldest
 38    api_versions: ClassVar[dict[str, list[type[APIVersionChange]]]] = {}
 39    api_version_header = "API-Version"
 40    default_api_version: str = ""
 41
 42    @cached_property
 43    def api_version(self) -> str:
 44        return self.get_api_version()
 45
 46    def get_api_version(self) -> str:
 47        version = ""
 48
 49        if version_name := self.request.headers.get(self.api_version_header, ""):
 50            version = version_name
 51        elif default_version := self.get_default_api_version():
 52            version = default_version
 53        else:
 54            raise ResponseException(
 55                Response(
 56                    f"Missing API version header '{self.api_version_header}'",
 57                    status_code=400,
 58                )
 59            )
 60
 61        if version in self.api_versions:
 62            return version
 63        else:
 64            raise ResponseException(
 65                Response(
 66                    f"Invalid API version '{version_name}'. Valid versions are: {', '.join(self.api_versions.keys())}",
 67                    status_code=400,
 68                )
 69            )
 70
 71    def get_default_api_version(self) -> str:
 72        # If this view has an api_key, use its version name
 73        if (api_key := getattr(self, "api_key", None)) and api_key.api_version:
 74            # If the API key has a version, use that
 75            return api_key.api_version
 76
 77        return self.default_api_version
 78
 79    def before_request(self) -> None:
 80        super().before_request()
 81        if self.request.content_type == "application/json":
 82            self.transform_request(self.request)
 83
 84    def after_response(self, response: Response) -> Response:
 85        response = super().after_response(response)
 86        if response.headers.get("Content-Type") == "application/json":
 87            self.transform_response(response)
 88
 89        # Put the API version on the response
 90        response.headers[self.api_version_header] = self.api_version
 91
 92        return response
 93
 94    def transform_request(self, request: Request) -> None:
 95        request_changes = []
 96
 97        # Find the version being requested,
 98        # then get every change after that up to the latest
 99        changing = False
100        for version, changes in reversed(self.api_versions.items()):
101            if version == self.api_version:
102                changing = True
103
104            if changing:
105                request_changes.extend(changes)
106
107        if not request_changes:
108            return
109
110        # Get the original request JSON
111        request_data = json.loads(request.body)
112
113        # Transform the request data for this version
114        for change in changes:
115            change().transform_request_forward(request, request_data)
116
117        # Update the request body with the transformed data
118        request._body = json.dumps(request_data).encode("utf-8")
119
120    def transform_response(self, response: Response) -> None:
121        response_changes = []
122
123        # Get the changes starting AFTER the current version
124        matching = False
125        for version, changes in reversed(self.api_versions.items()):
126            if matching:
127                response_changes.extend(changes)
128
129            if version == self.api_version:
130                matching = True
131
132        if not response_changes:
133            # No changes to apply, just return
134            return
135
136        # Get the original response JSON
137        response_data = json.loads(response.content)
138
139        for change in reversed(response_changes):
140            # Transform the response data for this version
141            change().transform_response_backward(response, response_data)
142
143        # Update the response body with the transformed data
144        response.content = json.dumps(response_data).encode("utf-8")