Объект управления HTTP-ответом

Для управления HTTP-ответами Rapidy использует объект Response.

from rapidy.http import Response

Для простых сценариев получения или сохранения данных rapidy.http.Response может быть не нужен, но если вы хотите гибко управлять ответом вашего HTTP-обработчика, то данный раздел расскажет о том как это можно делать.

Объект Response в Rapidy является оберткой над aiohttp.Response.

rapidy.http.Response в отличие от aiohttp.Response стал гораздо более вернеуровневым, у него появилась удобная сериализация данных.

Подробнее о aiohttp.Response можно узнать здесь.

Аттрибуты

body

body: Any | None = None — тело сообщения (может быть практически любым объектом).

Чтобы задать body, создайте объект Response(body=...) или используйте сеттер body существующего экземпляра.

from rapidy.http import Response, get

@get('/')
async def handler() -> Response:
    return Response(
        body={'hello': 'rapidy'},
    )

@get('/')
async def handler() -> ...:
    response = Response()
    response.body = {'hello': 'rapidy'}
    ...

@get('/')
async def handler(response: Response) -> ...:
    response.body = {'hello': 'rapidy'}
    ...

Для получения body воспользуйтесь геттером body, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response(body={'hello': 'rapidy'})
    body = response.body
    ...

Логика подготовки и преобразования данных завязана на аттрибуте content_type.

Объект типа bytes будет отправлен как тело без изменений.

Если вы хотите, чтобы работали флаги pydantic, такие как exclude_none и другие, переданный объект должен быть экземпляром pydantic.BaseModel.

text

text: Any | None = None — текстовое тело сообщения (может быть практически любым объектом).

Чтобы задать тело ответа как text, создайте объект Response(text=...) или используйте сеттер text существующего экземпляра.

from rapidy.http import Response, get

@get('/')
async def handler() -> Response:
    return Response(
        text = 'hello rapidy',
    )

@get('/')
async def handler() -> ...:
    response = Response()
    response.text = 'hello rapidy'
    ...

@get('/')
async def handler(response: Response) -> ...:
    response.text = 'hello rapidy'
    ...

Для получения тела ответа как text воспользуйтесь геттером text, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response(text='hello rapidy')
    body = response.text
    ...

Логика подготовки и преобразования данных завязана на аттрибуте content_type.

Объект типа str отправляется как тело без проверок (content_type="text/plain").

Если данные не str, применяется та же логика преобразования данных, что и для body.

content_type

content_type: str | ContentType | None = None — аттрибут позволяющий управлять заголовком Content-Type.

Заголовок Content-Type сообщает клиенту (браузеру, API-клиенту, другому серверу), какой тип данных содержится в теле HTTP-ответа.

Чтобы задать content_type, создайте объект Response(content_type=...) или используйте сеттер content_type существующего экземпляра.

from rapidy.http import get, Response, ContentType

@get('/')
async def handler() -> Response:
    return Response(
        body={'hello': 'rapidy!'},
        content_type=ContentType.json,
        # or
        content_type='application/json',
    )

@get('/')
async def handler() -> dict[str, str]:
    response = Response()

    response.content_type = ContentType.json
    # or
    response.content_type = 'application/json'

    return {'hello': 'rapidy!'}

@get('/')
async def handler(response: Response) -> dict[str, str]:
    response.content_type = ContentType.json
    # or
    response.content_type = 'application/json'

    return {'hello': 'rapidy!'}

Для получения content_type воспользуйтесь геттером content_type, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    content_type = response.content_type
    ...

Если указан content_type, переданные данные будут преобразованы в соответствии с ним.

Если content_type не указан - content_type будет определен автоматически в зависимости от типа данных которое отдает сервер.

content_type="application/json

content_type="application/json" — данные преобразуются в JSON с использованием jsonify(dumps=True) и кодируются в соответствии с charset.

from rapidy.http import get, Response, ContentType

@get('/')
async def handler() -> Response:
    return Response(
        body={'hello': 'rapidy!'},
        content_type=ContentType.json,
    )  # {"hello": "rapidy!"}

Если переданный объект является строкой Response(body="string"), то строка, согласно стандарту JSON, будет экранирована дважды:

from rapidy.http import get, Response, ContentType

@get('/')
async def handler() -> Response:
    return Response(
        body='hello rapidy!',
        content_type=ContentType.json,
    )  # "'hello rapidy!'"

content_type="text/*

content_type="text/*" (любой текстовый тип: text/plain, text/html и т. д.) - если данные имеют тип str, они отправляются без изменений. В противном случае они преобразуются в строку через jsonify(dumps=False).

from rapidy.http import get, Response, ContentType

@get('/')
async def handler() -> Response:
    return Response(
        text='hello rapidy!',
        content_type=ContentType.text_any,
    )  # "hello rapidy!"

Если после jsonify(dumps=False) объект не является строкой, он дополнительно преобразуется с помощью json_encoder, чтобы избежать двойного экранирования.

content_type - любой другой MIME-type.

Если данные имеют тип bytes, они отправляются без изменений. В противном случае они преобразуются в строку с использованием jsonify(dumps=True) и кодируются в соответствии с charset.

Если content_type не указан, он устанавливается автоматически:

body: dict | BaseModel | dataclass → content_type="application/json"

Response(body={"hello": "rapidy"})
Response(body=SomeModel(hello="rapidy"))

Response(body="string")
Response(body=SomeEnum.string)
Response(body=1)
Response(body=1.0)
Response(body=Decimal("1.0"))
Response(body=True)

body: Any → content_type="application/octet-stream"

Response(body=b'bytes')
Response(body=AnotherType())

status

status: int = 200 — HTTP-код ответа.

from rapidy.http import Response, get

@get('/')
async def handler() -> Response:
    return Response(
        status=201,
    )

set_status

Для установления status воспользуйтесь методом set_status, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    response.set_status(200)
    ...

@get('/')
async def handler(response: Response) -> ...:
    response.set_status(200)
    ...

headers

headers: Mapping[str, str] | None = None — дополнительные заголовки ответа.

from rapidy.http import Response, get

@get('/')
async def handler() -> Response:
    return Response(
        headers={'Some-Header': '123'},
    )

Для получения headers воспользуйтесь геттером headers, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    headers = response.headers
    ...

cookies

cookies: SimpleCookie | None = None - файлы cookie ответа для установки их браузером.

Для получения cookie воспользуйтесь геттером cookies, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    cookies = response.cookies
    ...

set_cookie

Для установления cookies воспользуйтесь методом set_cookie, объекта Response.

from rapidy.http import Response, get

@get('/')
async def handler() -> ...:
    response = Response()
    response.set_cookie('SomeCookie', 'SomeValue')
    ...

@get('/')
async def handler(response: Response) -> ...:
    response.set_cookie('SomeCookie', 'SomeValue')
    ...

del_cookie

Для удаления cookie воспользуйтесь методом del_cookie, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    response.del_cookie('SomeCookie')
    ...

charset

charset: str | Charset | None = 'utf-8' — кодировка, используемая для кодирования и декодирования данных.

Чтобы задать charset, создайте объект Response(charset=...) или используйте сеттер charset существующего экземпляра.

from rapidy.http import Response, get, Charset

@get('/')
async def handler() -> Response:
    return Response(
        charset=Charset.utf32,
        # or
        charset='utf32',
    )

@get('/')
async def handler(response: Response) -> ...:
    response.charset = Charset.utf32
    # or
    response.charset = 'utf32'

Для получения charset воспользуйтесь геттером charset, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    charset = response.charset
    ...

last_modified

last_modified: int | float | datetime.datetime | str | None = None — аттрибут отвечает за управление заголовком Last-Modified.

Для установления last_modified воспользуйтесь сеттером last_modified, объекта Response.

import datetime
from rapidy.http import get, Response

@get('/')
async def handler() -> ...:
    response = Response()
    # or
    response.last_modified = datetime.datetime(2024, 2, 24, 12, 0, 0, tzinfo=datetime.timezone.utc)
    # or
    response.last_modified = 'Wed, 21 Oct 2024 07:28:00 GMT'
    ...

@get('/')
async def handler(response: Response) -> ...:
    # or
    response.last_modified = datetime.datetime(2024, 2, 24, 12, 0, 0, tzinfo=datetime.timezone.utc)
    # or
    response.last_modified = 'Wed, 21 Oct 2024 07:28:00 GMT'
    ...

Для получения last_modified воспользуйтесь геттером last_modified, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    last_modified = response.last_modified
    ...

Полноценный пример построения HTTP-обработчика с использованием last_modified

from datetime import datetime, timezone
from typing import Annotated
from pydantic import BeforeValidator
from rapidy.http import Header, HTTPNotModified, Response, get

def parse_http_date(value: str) -> datetime:
    return datetime.strptime(value, '%a, %d %b %Y %H:%M:%S GMT').replace(tzinfo=timezone.utc)

IFModifiedSince = Annotated[
    datetime | None,
    BeforeValidator(parse_http_date),
    Header(None, alias='If-Modified-Since'),
]

@get('/')
async def handler(
    response: Response,
    if_modified_since: IFModifiedSince,
) -> str:
    last_mod_time = datetime(2024, 2, 24, 12, 0, 0, tzinfo=timezone.utc)

    # Check if the client sent the `If-Modified-Since` header
    if if_modified_since and if_modified_since >= last_mod_time:
        raise HTTPNotModified

    response.last_modified = last_mod_time  # Set the last modified date
    return 'success'

etag

etag: ETag | str - аттрибут отвечает за управление заголовком Etag.

Для установления etag воспользуйтесь сеттером etag, объекта Response.

from rapidy.http import get, Response

@get('/')
async def handler() -> ...:
    response = Response()
    response.etag = '33a64df551425fcc55e4d42a148795d9f25f89d4'
    ...

@get('/')
async def handler(response: Response) -> ...:
    response.etag = '33a64df551425fcc55e4d42a148795d9f25f89d4'
    ...

Для получения etag воспользуйтесь геттером etag, объекта Response.

...
@get('/')
async def handler() -> ...:
    response = Response()
    etag = response.etag
    ...

Полноценный пример построения HTTP-обработчика с использованием etag.

import hashlib
from rapidy.http import Header, HTTPNotModified, Response, get

@get('/')
async def handler(
    response: Response,
    if_none_match: str | None = Header(None, alias='If-None-Match'),
) -> str:
    content = 'success'

    # Generate ETag based on content
    etag_value = hashlib.md5(content.encode()).hexdigest()

    # Check If-None-Match
    if if_none_match and if_none_match == etag_value:
        raise HTTPNotModified

    response.etag = etag_value

    return content

include

include: set[str] | dict[str, Any] | None = None — параметр include из Pydantic, указывающий, какие поля включать.

from pydantic import BaseModel, Field
from rapidy.http import Response, get

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    another_value: str = Field('another_data', alias='someAnotherValue')

@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        include={'value'},
    )  # {'someValue': 'data'}

exclude

exclude: set[str] | dict[str, Any] | None = None — параметр exclude из Pydantic, указывающий, какие поля исключать.

from pydantic import BaseModel, Field
from rapidy.http import Response, get

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    another_value: str = Field('another_data', alias='someAnotherValue')

@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        exclude={'value'},
    )  # {'someAnotherValue': 'another_data'}

by_alias

by_alias: bool = True — параметр by_alias из Pydantic, определяющий, использовать ли псевдонимы атрибутов при сериализации.

from pydantic import BaseModel, Field
from rapidy.http import get, Response

class Result(BaseModel):
    value: str = Field('data', alias='someValue')

@get('/',)
async def handler() -> Response:
    return Response(
        Result(),
        by_alias=True,  # <-- default
    )  # {"someValue": "data"}

...
@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        by_alias=False,
    )  # {"value": "data"}

exclude_unset

exclude_unset: bool = False — параметр exclude_unset из Pydantic, исключающий поля, неявно установленные в значение по умолчанию.

from pydantic import BaseModel, Field
from rapidy.http import Response, get

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    another_value: str = Field('another_data', alias='someAnotherValue')

@get('/')
async def handler() -> Response:
    return Response(
        Result(someAnotherValue='new_data'),
        exclude_unset=False,  # <-- default
    )  # {"someValue": "data", "someAnotherValue": "new_data"}

...
@get('/')
async def handler() -> Response:
    return Response(
        Result(someAnotherValue='new_data'),
        exclude_unset=True,
    )  # {"someAnotherValue": "new_data"}

exclude_defaults

exclude_defaults: bool = False — параметр exclude_defaults из Pydantic, исключающий поля, имеющие значение по умолчанию, даже если они были явно заданы.

from pydantic import BaseModel, Field
from rapidy.http import Response, get

class Result(BaseModel):
    value: str = Field('data', alias='someValue')

@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        exclude_defaults=False,  # <-- default
    )  # {"value": "data"}

...
@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        exclude_defaults=True,
    )  # {}

exclude_none

exclude_none: bool = False — параметр exclude_none из Pydantic, исключающий из вывода поля со значением None.

from pydantic import BaseModel, Field
from rapidy.http import Response, get

class Result(BaseModel):
    value: str = Field('data', alias='someValue')
    none_value: None = None

@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        exclude_none=False,  # <-- default
    )  # {"someValue": "data", "none_value": null}

...
@get('/')
async def handler() -> Response:
    return Response(
        Result(),
        exclude_none=True,
    )  # {"someValue": "data"}

custom_encoder

custom_encoder: Callable | None = None — параметр custom_encoder из Pydantic, позволяющий задать пользовательский кодировщик.

json_encoder

json_encoder: Callable = json.dumps — функция, принимающая объект и возвращающая его JSON-представление.

from typing import Any
from rapidy.http import Response, get

def custom_encoder(obj: Any) -> str:
    ...

@get('/')
async def handler() -> Response:
    return Response(
        body={'hello': 'rapidy!'},  # will be converted to a string by Rapidy's internal tools
        json_encoder=custom_encoder,  # Converts the obtained string above into a JSON object using the `custom_encoder` function
    )