from __future__ import annotations
import itertools
from typing import TYPE_CHECKING, Any, Literal
from urllib.parse import urlencode
from litestar.constants import REDIRECT_ALLOWED_MEDIA_TYPES, REDIRECT_STATUS_CODES
from litestar.datastructures import MultiDict
from litestar.enums import MediaType
from litestar.exceptions import ImproperlyConfiguredException
from litestar.response.base import ASGIResponse, Response
from litestar.status_codes import HTTP_302_FOUND
from litestar.utils import url_quote
from litestar.utils.helpers import get_enum_string_value
if TYPE_CHECKING:
from collections.abc import Iterable, Mapping, Sequence
from litestar.background_tasks import BackgroundTask, BackgroundTasks
from litestar.connection import Request
from litestar.datastructures import Cookie
from litestar.types import ResponseCookies, ResponseHeaders, TypeEncodersMap
__all__ = (
"ASGIRedirectResponse",
"Redirect",
)
RedirectStatusType = Literal[301, 302, 303, 307, 308]
"""Acceptable status codes for redirect responses."""
[docs]
class ASGIRedirectResponse(ASGIResponse):
"""A low-level ASGI redirect response class."""
[docs]
def __init__(
self,
path: str | bytes,
media_type: str | None = None,
status_code: RedirectStatusType | None = None,
headers: dict[str, Any] | None = None,
background: BackgroundTask | BackgroundTasks | None = None,
body: bytes | str = b"",
content_length: int | None = None,
cookies: Iterable[Cookie] | None = None,
encoding: str = "utf-8",
is_head_response: bool = False,
) -> None:
headers = {**(headers or {}), "location": url_quote(path)}
media_type = media_type or MediaType.TEXT
status_code = status_code or HTTP_302_FOUND
if status_code not in REDIRECT_STATUS_CODES:
raise ImproperlyConfiguredException(
f"{status_code} is not a valid for this response. "
f"Redirect responses should have one of "
f"the following status codes: {', '.join([str(s) for s in REDIRECT_STATUS_CODES])}"
)
if media_type not in REDIRECT_ALLOWED_MEDIA_TYPES:
raise ImproperlyConfiguredException(
f"{media_type} media type is not supported yet. "
f"Media type should be one of "
f"the following values: {', '.join([str(s) for s in REDIRECT_ALLOWED_MEDIA_TYPES])}"
)
super().__init__(
status_code=status_code,
headers=headers,
media_type=media_type,
background=background,
is_head_response=is_head_response,
encoding=encoding,
cookies=cookies,
content_length=content_length,
body=body,
)
[docs]
class Redirect(Response[Any]):
"""A redirect response."""
__slots__ = ("url",)
[docs]
def __init__(
self,
path: str,
*,
background: BackgroundTask | BackgroundTasks | None = None,
cookies: ResponseCookies | None = None,
encoding: str = "utf-8",
headers: ResponseHeaders | None = None,
media_type: str | MediaType | None = None,
status_code: RedirectStatusType | None = None,
type_encoders: TypeEncodersMap | None = None,
query_params: Mapping[str, str | Sequence[str]] | MultiDict | None = None,
) -> None:
"""Initialize the response.
Args:
path: A path to redirect to.
background: A background task or tasks to be run after the response is sent.
cookies: A list of :class:`Cookie <.datastructures.Cookie>` instances to be set under the response
``Set-Cookie`` header.
encoding: The encoding to be used for the response headers.
headers: A string keyed dictionary of response headers. Header keys are insensitive.
media_type: A value for the response ``Content-Type`` header.
status_code: An HTTP status code. The status code should be one of 301, 302, 303, 307 or 308,
otherwise an exception will be raised.
type_encoders: A mapping of types to callables that transform them into types supported for serialization.
query_params: A dictionary of values from which the request's query will be generated.
Raises:
ImproperlyConfiguredException: Either if status code is not a redirect status code or media type is not
supported.
"""
if query_params is None:
self.url = path
elif isinstance(query_params, MultiDict):
# We can't use MultiDictMixin.dict() because it's not deterministic
query_params_dict = {k: query_params.getall(k) for k in query_params}
self.url = f"{path}?{urlencode(query_params_dict, doseq=True)}"
else:
self.url = f"{path}?{urlencode(query_params, doseq=True)}"
if status_code is None:
status_code = HTTP_302_FOUND
super().__init__(
background=background,
content=b"",
cookies=cookies,
encoding=encoding,
headers=headers,
media_type=media_type,
status_code=status_code,
type_encoders=type_encoders,
)
[docs]
def to_asgi_response(
self,
request: Request,
*,
background: BackgroundTask | BackgroundTasks | None = None,
cookies: Iterable[Cookie] | None = None,
headers: dict[str, str] | None = None,
is_head_response: bool = False,
media_type: MediaType | str | None = None,
status_code: int | None = None,
type_encoders: TypeEncodersMap | None = None,
) -> ASGIResponse:
headers = {**headers, **self.headers} if headers is not None else self.headers
cookies = self.cookies if cookies is None else itertools.chain(self.cookies, cookies)
media_type = get_enum_string_value(self.media_type or media_type or MediaType.TEXT)
return ASGIRedirectResponse(
path=self.url,
background=self.background or background,
body=b"",
content_length=None,
cookies=cookies,
encoding=self.encoding,
headers=headers,
is_head_response=is_head_response,
media_type=media_type,
status_code=self.status_code or status_code, # type:ignore[arg-type]
)
Litestar