JSONResponse class

ravyn.JSONResponse

JSONResponse(
    content,
    status_code=HTTP_200_OK,
    headers=None,
    media_type=None,
    background=None,
    encoders=None,
    passthrough_body_types=None,
)

Bases: ORJSONTransformMixin, BaseJSONResponse

An alternative to JSONResponse and performance wise, faster.

In the same way the JSONResponse is used, so is the ORJSONResponse.

Source code in lilya/responses.py

def __init__(
    self,
    content: Any,
    status_code: int = status.HTTP_200_OK,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: Task | None = None,
    encoders: Sequence[Encoder | type[Encoder]] | None = None,
    passthrough_body_types: tuple[type, ...] | None = None,
) -> None:
    super().__init__(
        content=content,
        status_code=status_code,
        headers=headers,
        media_type=media_type,
        background=background,
        encoders=encoders,
        passthrough_body_types=passthrough_body_types,
    )

media_type class-attribute instance-attribute

media_type = JSON

status_code class-attribute instance-attribute

status_code = None

charset class-attribute instance-attribute

charset = 'utf-8'

passthrough_body_types class-attribute instance-attribute

passthrough_body_types = (bytes,)

headers instance-attribute

headers

deduce_media_type_from_body class-attribute instance-attribute

deduce_media_type_from_body = deduce_media_type_from_body

cleanup_handler class-attribute instance-attribute

cleanup_handler = None

background instance-attribute

background = background

cookies instance-attribute

cookies = cookies

encoders instance-attribute

encoders = [
    (encoder() if isclass(encoder) else encoder)
    for encoder in (encoders or _empty)
]

async_content instance-attribute

async_content = content

body instance-attribute

body = make_response(content)

encoded_headers property

encoded_headers

raw_headers class-attribute instance-attribute

raw_headers = encoded_headers

execute_cleanup_handler async

execute_cleanup_handler()

Source code in lilya/responses.py

async def execute_cleanup_handler(self) -> None:
    if self.cleanup_handler is None:
        return
    res = self.cleanup_handler()
    if isawaitable(res):
        await res

resolve_async_content async

resolve_async_content()

Source code in lilya/responses.py

async def resolve_async_content(self) -> None:
    if getattr(self, "async_content", None) is not None:
        self.body = self.make_response(await self.async_content)
        self.async_content = None
        if (
            HeaderHelper.has_body_message(self.status_code)
            and "content-length" not in self.headers
        ):
            self.headers["content-length"] = str(len(self.body))
        if self.deduce_media_type_from_body:
            if self.deduce_media_type_from_body == "force" or self.media_type is None:
                self.media_type = self.find_media_type()
                self.headers["content-type"] = HeaderHelper.get_content_type(
                    charset=self.charset, media_type=self.media_type
                )

find_media_type

find_media_type()

Source code in lilya/responses.py

def find_media_type(self) -> str:
    require_magic()
    return magic.from_buffer(self.body[:2048], mime=True) or self.media_type or MediaType.OCTET

with_transform_kwargs classmethod

with_transform_kwargs(params)

Source code in lilya/responses.py

@classmethod
@contextlib.contextmanager
def with_transform_kwargs(cls, params: dict | None, /) -> Generator[None, None, None]:
    token = RESPONSE_TRANSFORM_KWARGS.set(params)
    try:
        yield
    finally:
        RESPONSE_TRANSFORM_KWARGS.reset(token)

transform classmethod

transform(value)

The transformation of the data being returned (simplify operation).

Supports all the default encoders from Lilya and custom from Ravyn.

Source code in ravyn/responses/mixins.py

@classmethod
def transform(cls, value: Any) -> Any:
    """
    The transformation of the data being returned (simplify operation).

    Supports all the default encoders from Lilya and custom from Ravyn.
    """
    transform_kwargs = RESPONSE_TRANSFORM_KWARGS.get()
    if transform_kwargs is None:
        transform_kwargs = {}
    else:
        transform_kwargs = transform_kwargs.copy()
    transform_kwargs.setdefault(
        "json_encode_fn",
        partial(
            orjson.dumps, option=orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_OMIT_MICROSECONDS
        ),
    )
    transform_kwargs.setdefault("post_transform_fn", orjson.loads)

    with cls.with_transform_kwargs(transform_kwargs):
        return super().transform(value)

make_headers

make_headers(content_headers=None)

Initializes the headers based on RFC specifications by setting appropriate conditions and restrictions.

PARAMETER	DESCRIPTION
content_headers	Additional headers to include (default is None). TYPE: Union[Mapping[str, str], Dict[str, str], None] DEFAULT: None

Source code in lilya/responses.py

def make_headers(
    self, content_headers: Mapping[str, str] | dict[str, str] | None = None
) -> None:
    """
    Initializes the headers based on RFC specifications by setting appropriate conditions and restrictions.

    Args:
        content_headers (Union[Mapping[str, str], Dict[str, str], None], optional): Additional headers to include (default is None).
    """
    headers: dict[str, str] = {} if content_headers is None else content_headers  # type: ignore

    if HeaderHelper.has_entity_header_status(self.status_code):
        headers = HeaderHelper.remove_entity_headers(headers)
    if HeaderHelper.has_body_message(self.status_code):
        if getattr(self, "body", None) is not None:
            headers.setdefault("content-length", str(len(self.body)))

        # Populates the content type if exists and either a body was found or deduce_media_type_from_body was not force
        if (
            self.deduce_media_type_from_body != "force"
            or getattr(self, "body", None) is not None
        ):
            content_type = HeaderHelper.get_content_type(
                charset=self.charset, media_type=self.media_type
            )
            if content_type is not None:
                headers.setdefault("content-type", content_type)
    self.headers = Header(headers)

set_cookie

set_cookie(
    key,
    value="",
    *,
    path="/",
    domain=None,
    secure=False,
    max_age=None,
    expires=None,
    httponly=False,
    samesite="lax",
)

Sets a cookie in the response headers.

PARAMETER	DESCRIPTION
key	The name of the cookie. TYPE: str
value	The value of the cookie. TYPE: str DEFAULT: ''
path	The path for which the cookie is valid (default is '/'). TYPE: str DEFAULT: '/'
domain	The domain to which the cookie belongs. TYPE: Union[str, None] DEFAULT: None
secure	If True, the cookie should only be sent over HTTPS. TYPE: bool DEFAULT: False
max_age	The maximum age of the cookie in seconds. TYPE: Union[int, None] DEFAULT: None
expires	The expiration date of the cookie. TYPE: Union[Union[datetime, str, int], None] DEFAULT: None
httponly	If True, the cookie should only be accessible through HTTP. TYPE: bool DEFAULT: False
samesite	SameSite attribute of the cookie. TYPE: Literal['lax', 'strict', 'none'] DEFAULT: 'lax'

RAISES	DESCRIPTION
AssertionError	If samesite is not one of 'strict', 'lax', or 'none'.

Source code in lilya/responses.py

def set_cookie(
    self,
    key: str,
    value: str = "",
    *,
    path: str = "/",
    domain: str | None = None,
    secure: bool = False,
    max_age: int | None = None,
    expires: datetime | str | int | None = None,
    httponly: bool = False,
    samesite: Literal["lax", "strict", "none"] = "lax",
) -> None:
    """
    Sets a cookie in the response headers.

    Args:
        key (str): The name of the cookie.
        value (str, optional): The value of the cookie.
        path (str, optional): The path for which the cookie is valid (default is '/').
        domain (Union[str, None], optional): The domain to which the cookie belongs.
        secure (bool, optional): If True, the cookie should only be sent over HTTPS.
        max_age (Union[int, None], optional): The maximum age of the cookie in seconds.
        expires (Union[Union[datetime, str, int], None], optional): The expiration date of the cookie.
        httponly (bool, optional): If True, the cookie should only be accessible through HTTP.
        samesite (Literal["lax", "strict", "none"], optional): SameSite attribute of the cookie.

    Raises:
        AssertionError: If samesite is not one of 'strict', 'lax', or 'none'.
    """
    cookie: http.cookies.BaseCookie[str] = http.cookies.SimpleCookie()
    cookie[key] = value
    if max_age is not None:
        cookie[key]["max-age"] = max_age
    if expires is not None:
        if isinstance(expires, datetime):
            cookie[key]["expires"] = format_datetime(expires, usegmt=True)
        else:
            cookie[key]["expires"] = expires
    if path is not None:
        cookie[key]["path"] = path
    if domain is not None:
        cookie[key]["domain"] = domain
    if secure:
        cookie[key]["secure"] = True
    if httponly:
        cookie[key]["httponly"] = True
    if samesite is not None:
        assert samesite.lower() in [
            "strict",
            "lax",
            "none",
        ], "samesite must be either 'strict', 'lax' or 'none'"
        cookie[key]["samesite"] = samesite
    cookie_val = cookie.output(header="").strip()
    self.headers.add("set-cookie", cookie_val)

delete_cookie

delete_cookie(
    key,
    path="/",
    domain=None,
    secure=False,
    httponly=False,
    samesite="lax",
)

Deletes a cookie in the response headers by setting its max age and expiration to 0.

PARAMETER	DESCRIPTION
key	The name of the cookie to delete. TYPE: str
path	The path for which the cookie is valid (default is '/'). TYPE: str DEFAULT: '/'
domain	The domain to which the cookie belongs. TYPE: Union[str, None] DEFAULT: None
secure	If True, the cookie should only be sent over HTTPS. TYPE: bool DEFAULT: False
httponly	If True, the cookie should only be accessible through HTTP. TYPE: bool DEFAULT: False
samesite	SameSite attribute of the cookie. TYPE: Literal['lax', 'strict', 'none'] DEFAULT: 'lax'

Source code in lilya/responses.py

def delete_cookie(
    self,
    key: str,
    path: str = "/",
    domain: str | None = None,
    secure: bool = False,
    httponly: bool = False,
    samesite: Literal["lax", "strict", "none"] = "lax",
) -> None:
    """
    Deletes a cookie in the response headers by setting its max age and expiration to 0.

    Args:
        key (str): The name of the cookie to delete.
        path (str, optional): The path for which the cookie is valid (default is '/').
        domain (Union[str, None], optional): The domain to which the cookie belongs.
        secure (bool, optional): If True, the cookie should only be sent over HTTPS.
        httponly (bool, optional): If True, the cookie should only be accessible through HTTP.
        samesite (Literal["lax", "strict", "none"], optional): SameSite attribute of the cookie.
    """
    self.set_cookie(
        key,
        max_age=0,
        expires=0,
        path=path,
        domain=domain,
        secure=secure,
        httponly=httponly,
        samesite=samesite,
    )

message

message(prefix)

Source code in lilya/responses.py

def message(self, prefix: str) -> dict[str, Any]:
    return {
        "type": prefix + "http.response.start",
        "status": self.status_code,
        # some tests add headers dirty and assume a list
        "headers": self.headers,
    }

make_response

make_response(content)

Source code in ravyn/responses/encoders.py

def make_response(self, content: Any) -> bytes:
    new_params = RESPONSE_TRANSFORM_KWARGS.get()
    if new_params:
        new_params = new_params.copy()
    else:
        new_params = {}
    new_params.setdefault(
        "json_encode_fn",
        partial(
            orjson.dumps,
            option=orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_OMIT_MICROSECONDS,
        ),
    )
    with self.with_transform_kwargs(new_params):
        return super().make_response(content)