Source code for dmr.renderers

import abc
from collections.abc import Callable, Mapping
from http import HTTPStatus
from typing import TYPE_CHECKING, Any, ClassVar

from typing_extensions import override

from dmr.exceptions import NotAcceptableError, ResponseSchemaError
from dmr.internal.json import JsonModule, NativeJson
from dmr.metadata import EndpointMetadata, ResponseSpec, ResponseSpecProvider
from dmr.parsers import (
    JsonParser,
    Parser,
    _NoOpParser,  # pyright: ignore[reportPrivateUsage]
)

if TYPE_CHECKING:
    from dmr.controller import Controller
    from dmr.serializer import BaseSerializer




class Renderer(ResponseSpecProvider):
    """
    Base class for all renderer types.

    Subclass it to implement your own renderers.
    """

    __slots__ = ()

    content_type: str
    """
    Content-Type that this renderer works with.

    Must be defined for all subclasses.
    """

    streaming: ClassVar[bool] = False
    """Whether or not this renderer is used for streaming responses."""



    @abc.abstractmethod
    def render(
        self,
        to_serialize: Any,
        serializer_hook: Callable[[Any], Any],
    ) -> bytes:
        """Function to be called on object serialization."""


    @property
    @abc.abstractmethod
    def validation_parser(self) -> Parser:
        """
        Returns a parser that can parse what this renderer rendered.

        Why? Because when ``validate_responses`` is ``True``,
        we parse the response body once again to see if it fits the schema.

        That's why all renderers must know how to unparse its results.
        """
        raise NotImplementedError



    @override
    def provide_response_specs(
        self,
        metadata: EndpointMetadata,
        controller_cls: type['Controller[BaseSerializer]'],
        existing_responses: Mapping[HTTPStatus, ResponseSpec],
    ) -> list[ResponseSpec]:
        """Provides responses that can happen when data can't be rendered."""
        # This is technically not renderer's response, but it is the closest.
        response_validation = (
            self._add_new_response(
                ResponseSpec(
                    return_type=controller_cls.error_model,
                    status_code=ResponseSchemaError.status_code,
                    description=(
                        'Raised when returned response does not '
                        'match the response schema'
                    ),
                ),
                existing_responses,
            )
            # When validation is disabled, `ResponseSchemaError` can't happen.
            if metadata.validate_responses
            else []
        )
        return [
            *response_validation,
            *self._add_new_response(
                # When we face wrong `Accept` header, we raise 406 error:
                ResponseSpec(
                    return_type=controller_cls.error_model,
                    status_code=NotAcceptableError.status_code,
                    description=(
                        'Raised when provided `Accept` header '
                        'cannot be satisfied'
                    ),
                ),
                existing_responses,
            ),
        ]






class JsonRenderer(Renderer):
    """
    Fallback implementation of a json renderer.

    Only is used when ``msgspec`` is not installed.

    .. warning::

        It is not recommended to be used directly.
        It is slow and has less features.
        We won't add any complex objects support to this renderer.

    Alternative ``json`` implementations can be provided.
    See :ref:`alternative-json` for more info.
    """

    __slots__ = (
        '_json_module',
        'content_type',
    )

    def __init__(
        self,
        content_type: str = 'application/json',
        *,
        json_module: JsonModule = NativeJson,
    ) -> None:
        """Init the renderer with all defaults."""
        self.content_type = content_type
        self._json_module = json_module
        # Sanity check:
        assert self._json_module.dumps, (  # type: ignore[truthy-function]  # noqa: S101
            'Passed json module does not have `.dumps` method'
        )



    @override
    def render(
        self,
        to_serialize: Any,
        serializer_hook: Callable[[Any], Any],
    ) -> bytes:
        """
        Encode a value into JSON bytestring.

        Args:
            to_serialize: Value to encode.
            serializer_hook: Callable to support non-natively supported types.

        Returns:
            JSON as bytes.

        """
        # msgspec returns `bytes`, we prefer to use `bytes` by default
        # and not to create extra strings when not needed in "fast" mode.
        # We don't really care about raw json implementation. It is a fallback.
        return self._json_module.dumps(to_serialize, default=serializer_hook)


    @property
    @override
    def validation_parser(self) -> JsonParser:
        """Regular json parser can parse this."""
        return JsonParser(json_module=self._json_module)





class FileRenderer(Renderer):
    """
    Renders any file.

    Works with any files and any content types.

    .. warning::

        Works with any content type by default,
        so it must be an only renderer for the endpoint.

    """

    __slots__ = ('content_type',)

    def __init__(self, content_type: str = '*/*') -> None:
        """Users can customize content types that this renderer works with."""
        self.content_type = content_type



    @override
    def render(
        self,
        to_serialize: Any,
        serializer_hook: Callable[[Any], Any],
    ) -> bytes:
        """Render a file."""
        raise NotImplementedError(
            'FileRenderer.render() must not be called, '
            'instead return a FileResponse directly',
        )


    @property
    @override
    def validation_parser(self) -> _NoOpParser:
        """Since there's nothing to parse, we return a no-op."""
        return _NoOpParser(self.content_type)