Skip to content

gitea.utils.response

response

Utility functions for processing HTTP responses.

Functions:

gitea.utils.response.process_response 

process_response(
    response: Response, default: T | None = None
) -> tuple[Any, int]

Process a synchronous HTTP response.

Parameters:

Name Type Description Default
response Response

The HTTP response object.

 required
default T | None

The default value to return if parsing fails.

 None

Returns:

Type Description
tuple[Any, int]

A tuple containing the response data and status code.
Source code in src/gitea/utils/response.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def process_response[T](response: Response, default: T | None = None) -> tuple[Any, int]:
    """Process a synchronous HTTP response.

    Args:
        response: The HTTP response object.
        default: The default value to return if parsing fails.

    Returns:
        A tuple containing the response data and status code.

    """
    status_code = response.status_code
    if status_code == 204:  # noqa: PLR2004
        data = default
    elif 200 <= status_code < 300:  # noqa: PLR2004
        if not response.content:
            data = default
        else:
            try:
                data = response.json()
            except ValueError as e:
                logger.error("Failed to parse JSON response: %s", e)
                data = default
    else:
        data = default
    return data, cast(int, status_code)

gitea.utils.response.process_async_response async 

process_async_response(
    response: ClientResponse, default: T | None = None
) -> tuple[Any, int]

Process an asynchronous HTTP response.

Parameters:

Name Type Description Default
response ClientResponse

The asynchronous HTTP response object.

 required
default T | None

The default value to return if parsing fails.

 None

Returns:

Type Description
tuple[Any, int]

A tuple containing the response data and status code.
Source code in src/gitea/utils/response.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def process_async_response[T](response: ClientResponse, default: T | None = None) -> tuple[Any, int]:
    """Process an asynchronous HTTP response.

    Args:
        response: The asynchronous HTTP response object.
        default: The default value to return if parsing fails.

    Returns:
        A tuple containing the response data and status code.

    """
    status_code = response.status
    if status_code == 204:  # noqa: PLR2004
        data = default
    elif 200 <= status_code < 300:  # noqa: PLR2004
        body = await response.read()
        if not body:
            data = default
        else:
            try:
                data = json.loads(body)
            except (ValueError, UnicodeDecodeError) as e:
                logger.error("Failed to parse JSON response: %s", e)
                data = default
    else:
        data = default
    return data, status_code

gitea.utils.response.process_text_response 

process_text_response(
    response: Response, default: str = ""
) -> tuple[str, int]

Process a synchronous HTTP response whose body is text rather than JSON.

A few endpoints answer with a file rather than with a document - the logs of an Actions job are the log itself - so there is nothing to parse and nothing to fall back to when parsing fails.

The body is decoded as UTF-8 rather than through response.text, which guesses the encoding when the response declares none: a log blob served as application/octet-stream would be decoded by that guess, and the same bytes would then reach a caller differently depending on what the guess was. Gitea writes these as UTF-8, and the asynchronous path decodes them the same way, so both clients hand back the same text. A byte that is not valid UTF-8 is replaced rather than raising, since a log is worth reading even where one line of it is malformed.

Parameters:

Name Type Description Default
response Response

The HTTP response object.

 required
default str

The value to return when the response carries no body.

 ''

Returns:

Type Description
tuple[str, int]

A tuple containing the response text and status code.
Source code in src/gitea/utils/response.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def process_text_response(response: Response, default: str = "") -> tuple[str, int]:
    """Process a synchronous HTTP response whose body is text rather than JSON.

    A few endpoints answer with a file rather than with a document - the logs of
    an Actions job are the log itself - so there is nothing to parse and nothing
    to fall back to when parsing fails.

    The body is decoded as UTF-8 rather than through `response.text`, which
    guesses the encoding when the response declares none: a log blob served as
    `application/octet-stream` would be decoded by that guess, and the same
    bytes would then reach a caller differently depending on what the guess
    was. Gitea writes these as UTF-8, and the asynchronous path decodes them the
    same way, so both clients hand back the same text. A byte that is not valid
    UTF-8 is replaced rather than raising, since a log is worth reading even
    where one line of it is malformed.

    Args:
        response: The HTTP response object.
        default: The value to return when the response carries no body.

    Returns:
        A tuple containing the response text and status code.

    """
    status_code = response.status_code
    if 200 <= status_code < 300 and response.content:  # noqa: PLR2004
        return response.content.decode("utf-8", errors="replace"), cast(int, status_code)
    return default, cast(int, status_code)

gitea.utils.response.process_async_text_response async 

process_async_text_response(
    response: ClientResponse, default: str = ""
) -> tuple[str, int]

Process an asynchronous HTTP response whose body is text rather than JSON.

Decoded as process_text_response decodes it, so a caller reading the logs of a job gets the same text from either client.

Parameters:

Name Type Description Default
response ClientResponse

The asynchronous HTTP response object.

 required
default str

The value to return when the response carries no body.

 ''

Returns:

Type Description
tuple[str, int]

A tuple containing the response text and status code.
Source code in src/gitea/utils/response.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def process_async_text_response(response: ClientResponse, default: str = "") -> tuple[str, int]:
    """Process an asynchronous HTTP response whose body is text rather than JSON.

    Decoded as `process_text_response` decodes it, so a caller reading the logs
    of a job gets the same text from either client.

    Args:
        response: The asynchronous HTTP response object.
        default: The value to return when the response carries no body.

    Returns:
        A tuple containing the response text and status code.

    """
    status_code = response.status
    if 200 <= status_code < 300:  # noqa: PLR2004
        body = await response.read()
        if body:
            return body.decode("utf-8", errors="replace"), status_code
    return default, status_code

gitea.utils.response.process_binary_response 

process_binary_response(
    response: Response, default: bytes = b""
) -> tuple[bytes, int]

Process a synchronous HTTP response whose body is a file rather than a document.

An Actions artifact is a zip archive, so there is nothing to parse and nothing to decode: handing back the bytes is the whole of it. Decoding them as text - as the log endpoints are decoded - would replace every byte that is not valid UTF-8 and produce an archive that no longer opens, which is why this exists alongside process_text_response rather than reusing it.

Parameters:

Name Type Description Default
response Response

The HTTP response object.

 required
default bytes

The value to return when the response carries no body.

 b''

Returns:

Type Description
tuple[bytes, int]

A tuple containing the response body and status code.
Source code in src/gitea/utils/response.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def process_binary_response(response: Response, default: bytes = b"") -> tuple[bytes, int]:
    """Process a synchronous HTTP response whose body is a file rather than a document.

    An Actions artifact is a zip archive, so there is nothing to parse and
    nothing to decode: handing back the bytes is the whole of it. Decoding them
    as text - as the log endpoints are decoded - would replace every byte that is
    not valid UTF-8 and produce an archive that no longer opens, which is why
    this exists alongside `process_text_response` rather than reusing it.

    Args:
        response: The HTTP response object.
        default: The value to return when the response carries no body.

    Returns:
        A tuple containing the response body and status code.

    """
    status_code = response.status_code
    if 200 <= status_code < 300 and response.content:  # noqa: PLR2004
        return response.content, cast(int, status_code)
    return default, cast(int, status_code)

gitea.utils.response.process_async_binary_response async 

process_async_binary_response(
    response: ClientResponse, default: bytes = b""
) -> tuple[bytes, int]

Process an asynchronous HTTP response whose body is a file rather than a document.

Handed back as process_binary_response hands it back, so an artifact downloaded through either client is the same archive.

Parameters:

Name Type Description Default
response ClientResponse

The asynchronous HTTP response object.

 required
default bytes

The value to return when the response carries no body.

 b''

Returns:

Type Description
tuple[bytes, int]

A tuple containing the response body and status code.
Source code in src/gitea/utils/response.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
async def process_async_binary_response(response: ClientResponse, default: bytes = b"") -> tuple[bytes, int]:
    """Process an asynchronous HTTP response whose body is a file rather than a document.

    Handed back as `process_binary_response` hands it back, so an artifact
    downloaded through either client is the same archive.

    Args:
        response: The asynchronous HTTP response object.
        default: The value to return when the response carries no body.

    Returns:
        A tuple containing the response body and status code.

    """
    status_code = response.status
    if 200 <= status_code < 300:  # noqa: PLR2004
        body = await response.read()
        if body:
            return body, status_code
    return default, status_code