• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

MrThearMan / undine / 16081423623

04 Jul 2025 09:57PM UTC coverage: 97.685%. First build
16081423623

Pull #33

github

web-flow
Merge 6eb57167c into 784a68391
Pull Request #33: Add Subscriptions

1798 of 1841 branches covered (97.66%)

Branch coverage included in aggregate %.

1009 of 1176 new or added lines in 36 files covered. (85.8%)

26853 of 27489 relevant lines covered (97.69%)

8.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.32
/undine/http/utils.py
1
from __future__ import annotations
9✔
2

3
import json
9✔
4
from asyncio import iscoroutinefunction
9✔
5
from collections.abc import Awaitable, Callable
9✔
6
from functools import wraps
9✔
7
from http import HTTPStatus
9✔
8
from typing import TYPE_CHECKING, Any, TypeAlias, overload
9✔
9

10
from django.http import HttpRequest, HttpResponse
9✔
11
from django.http.request import MediaType
9✔
12
from django.shortcuts import render
9✔
13
from graphql import ExecutionResult
9✔
14

15
from undine.exceptions import (
9✔
16
    GraphQLMissingContentTypeError,
17
    GraphQLRequestDecodingError,
18
    GraphQLUnsupportedContentTypeError,
19
)
20
from undine.settings import undine_settings
9✔
21
from undine.typing import DjangoRequestProtocol, DjangoResponseProtocol
9✔
22

23
if TYPE_CHECKING:
24
    from collections.abc import Iterable
25

26
    from graphql import GraphQLError
27

28
    from undine.exceptions import GraphQLErrorGroup
29
    from undine.typing import RequestMethod
30

31
__all__ = [
9✔
32
    "HttpMethodNotAllowedResponse",
33
    "HttpUnsupportedContentTypeResponse",
34
    "decode_body",
35
    "get_preferred_response_content_type",
36
    "graphql_error_group_response",
37
    "graphql_error_response",
38
    "graphql_result_response",
39
    "load_json_dict",
40
    "parse_json_body",
41
    "render_graphiql",
42
    "require_graphql_request",
43
    "require_persisted_documents_request",
44
]
45

46

47
class HttpMethodNotAllowedResponse(HttpResponse):
9✔
48
    def __init__(self, allowed_methods: Iterable[RequestMethod]) -> None:
9✔
49
        msg = "Method not allowed"
9✔
50
        super().__init__(content=msg, status=HTTPStatus.METHOD_NOT_ALLOWED, content_type="text/plain; charset=utf-8")
9✔
51
        self["Allow"] = ", ".join(allowed_methods)
9✔
52

53

54
class HttpUnsupportedContentTypeResponse(HttpResponse):
9✔
55
    def __init__(self, supported_types: Iterable[str]) -> None:
9✔
56
        msg = "Server does not support any of the requested content types."
9✔
57
        super().__init__(content=msg, status=HTTPStatus.NOT_ACCEPTABLE, content_type="text/plain; charset=utf-8")
9✔
58
        self["Accept"] = ", ".join(supported_types)
9✔
59

60

61
def get_preferred_response_content_type(accepted: list[MediaType], supported: list[str]) -> str | None:
9✔
62
    """Get the first supported media type matching given accepted types."""
63
    for accepted_type in accepted:
9✔
64
        for supported_type in supported:
9✔
65
            if accepted_type.match(supported_type):
9✔
66
                return supported_type
9✔
67
    return None
9✔
68

69

70
def parse_json_body(body: bytes, charset: str = "utf-8") -> dict[str, Any]:
9✔
71
    """
72
    Parse JSON body.
73

74
    :param body: The body to parse.
75
    :param charset: The charset to decode the body with.
76
    :raises GraphQLDecodeError: If the body cannot be decoded.
77
    :return: The parsed JSON body.
78
    """
79
    decoded = decode_body(body, charset=charset)
9✔
80
    return load_json_dict(
9✔
81
        decoded,
82
        decode_error_msg="Could not load JSON body.",
83
        type_error_msg="JSON body should convert to a dictionary.",
84
    )
85

86

87
def decode_body(body: bytes, charset: str = "utf-8") -> str:
9✔
88
    """
89
    Decode body.
90

91
    :param body: The body to decode.
92
    :param charset: The charset to decode the body with.
93
    :raises GraphQLRequestDecodingError: If the body cannot be decoded.
94
    :return: The decoded body.
95
    """
96
    try:
9✔
97
        return body.decode(encoding=charset)
9✔
98
    except Exception as error:
9✔
99
        msg = f"Could not decode body with encoding '{charset}'."
9✔
100
        raise GraphQLRequestDecodingError(msg) from error
9✔
101

102

103
def load_json_dict(string: str, *, decode_error_msg: str, type_error_msg: str) -> dict[str, Any]:
9✔
104
    """
105
    Load JSON dict from string, raising GraphQL errors if decoding fails.
106

107
    :param string: The string to load.
108
    :param decode_error_msg: The error message to use if decoding fails.
109
    :param type_error_msg: The error message to use if the string is not a JSON object.
110
    :raises GraphQLRequestDecodingError: If decoding fails or the string is not a JSON object.
111
    :return: The loaded JSON dict.
112
    """
113
    try:
9✔
114
        data = json.loads(string)
9✔
115
    except Exception as error:
9✔
116
        raise GraphQLRequestDecodingError(decode_error_msg) from error
9✔
117

118
    if not isinstance(data, dict):
9✔
119
        raise GraphQLRequestDecodingError(type_error_msg)
9✔
120
    return data
9✔
121

122

123
def graphql_result_response(
9✔
124
    result: ExecutionResult,
125
    *,
126
    status: int = HTTPStatus.OK,
127
    content_type: str = "application/json",
128
) -> DjangoResponseProtocol:
129
    """Serialize the given execution result to an HTTP response."""
130
    content = json.dumps(result.formatted, separators=(",", ":"))
9✔
131
    return HttpResponse(content=content, status=status, content_type=content_type)  # type: ignore[return-value]
9✔
132

133

134
def graphql_error_response(
9✔
135
    error: GraphQLError,
136
    *,
137
    status: int = HTTPStatus.OK,
138
    content_type: str = "application/json",
139
) -> DjangoResponseProtocol:
140
    """Serialize the given GraphQL error to an HTTP response."""
141
    result = ExecutionResult(errors=[error])
9✔
142
    return graphql_result_response(result, status=status, content_type=content_type)
9✔
143

144

145
def graphql_error_group_response(
9✔
146
    error: GraphQLErrorGroup,
147
    *,
148
    status: int = HTTPStatus.OK,
149
    content_type: str = "application/json",
150
) -> DjangoResponseProtocol:
151
    """Serialize the given GraphQL error group to an HTTP response."""
152
    result = ExecutionResult(errors=list(error.flatten()))
9✔
153
    return graphql_result_response(result, status=status, content_type=content_type)
9✔
154

155

156
SyncViewIn: TypeAlias = Callable[[DjangoRequestProtocol], DjangoResponseProtocol]
9✔
157
AsyncViewIn: TypeAlias = Callable[[DjangoRequestProtocol], Awaitable[DjangoResponseProtocol]]
9✔
158

159
SyncViewOut: TypeAlias = Callable[[HttpRequest], HttpResponse]
9✔
160
AsyncViewOut: TypeAlias = Callable[[HttpRequest], Awaitable[HttpResponse]]
9✔
161

162

163
@overload
9✔
164
def require_graphql_request(func: SyncViewIn) -> SyncViewOut: ...
9✔
165

166

167
@overload
9✔
168
def require_graphql_request(func: AsyncViewIn) -> AsyncViewOut: ...
9✔
169

170

171
def require_graphql_request(func: SyncViewIn | AsyncViewIn) -> SyncViewOut | AsyncViewOut:
9✔
172
    """
173
    Perform various checks on the request to ensure it's suitable for GraphQL operations.
174
    Can also return early to display GraphiQL.
175
    """
176
    methods: list[RequestMethod] = ["GET", "POST"]
9✔
177

178
    def get_supported_types() -> list[str]:
9✔
179
        supported_types = ["application/graphql-response+json", "application/json"]
9✔
180
        if undine_settings.GRAPHIQL_ENABLED:
9✔
181
            supported_types.append("text/html")
9✔
182
        return supported_types
9✔
183

184
    if iscoroutinefunction(func):
9✔
185

186
        @wraps(func)
9✔
187
        async def wrapper(request: DjangoRequestProtocol) -> DjangoResponseProtocol | HttpResponse:
9✔
188
            if request.method not in methods:
9✔
189
                return HttpMethodNotAllowedResponse(allowed_methods=methods)
×
190

191
            supported_types = get_supported_types()
9✔
192
            media_type = get_preferred_response_content_type(accepted=request.accepted_types, supported=supported_types)
9✔
193
            if media_type is None:
9✔
194
                return HttpUnsupportedContentTypeResponse(supported_types=supported_types)
×
195

196
            if media_type == "text/html":
9✔
NEW
197
                return render_graphiql(request)  # type: ignore[arg-type]
×
198

199
            request.response_content_type = media_type
9✔
200
            return await func(request)
9✔
201

202
    else:
203

204
        @wraps(func)
9✔
205
        def wrapper(request: DjangoRequestProtocol) -> DjangoResponseProtocol | HttpResponse:
9✔
206
            if request.method not in methods:
9✔
207
                return HttpMethodNotAllowedResponse(allowed_methods=methods)
9✔
208

209
            supported_types = get_supported_types()
9✔
210
            media_type = get_preferred_response_content_type(accepted=request.accepted_types, supported=supported_types)
9✔
211
            if media_type is None:
9✔
212
                return HttpUnsupportedContentTypeResponse(supported_types=supported_types)
9✔
213

214
            if media_type == "text/html":
9✔
215
                return render_graphiql(request)  # type: ignore[arg-type]
9✔
216

217
            request.response_content_type = media_type
9✔
218
            return func(request)  # type: ignore[return-value]
9✔
219

220
    return wrapper  # type: ignore[return-value]
9✔
221

222

223
def require_persisted_documents_request(func: SyncViewIn) -> SyncViewOut:
9✔
224
    """Perform various checks on the request to ensure that it's suitable for registering persisted documents."""
225
    content_type: str = "application/json"
9✔
226
    methods: list[RequestMethod] = ["POST"]
9✔
227

228
    @wraps(func)
9✔
229
    def wrapper(request: DjangoRequestProtocol) -> DjangoResponseProtocol | HttpResponse:
9✔
230
        if request.method not in methods:
9✔
231
            return HttpMethodNotAllowedResponse(allowed_methods=methods)
×
232

233
        media_type = get_preferred_response_content_type(accepted=request.accepted_types, supported=[content_type])
9✔
234
        if media_type is None:
9✔
235
            return HttpUnsupportedContentTypeResponse(supported_types=[content_type])
9✔
236

237
        request.response_content_type = media_type
9✔
238

239
        if request.content_type is None:  # pragma: no cover
240
            return graphql_error_response(
241
                error=GraphQLMissingContentTypeError(),
242
                status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
243
                content_type=media_type,
244
            )
245

246
        if not MediaType(request.content_type).match(content_type):
9✔
247
            return graphql_error_response(
9✔
248
                error=GraphQLUnsupportedContentTypeError(content_type=request.content_type),
249
                status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
250
                content_type=media_type,
251
            )
252

253
        return func(request)
9✔
254

255
    return wrapper  # type: ignore[return-value]
9✔
256

257

258
def render_graphiql(request: HttpRequest) -> HttpResponse:
9✔
259
    """Render GraphiQL."""
260
    return render(request, "undine/graphiql.html", context=get_graphiql_context())
9✔
261

262

263
def get_graphiql_context() -> dict[str, Any]:
9✔
264
    """Get the GraphiQL context."""
265
    return {
9✔
266
        "http_path": undine_settings.GRAPHQL_PATH,
267
        "ws_path": undine_settings.WEBSOCKET_PATH,
268
        "importmap": get_importmap(),
269
        # Note that changing the versions here will break integrity checks! Regenerate: https://www.srihash.org/
270
        "graphiql_css": "https://esm.sh/graphiql@5.0.3/dist/style.css",
271
        "explorer_css": "https://esm.sh/@graphiql/plugin-explorer@5.0.0/dist/style.css",
272
        "graphiql_css_integrity": "sha384-lixdMC836B3JdnFulLFPKjIN0gr85IffJ5qBAoYmKxoeNXlkn+JgibHqHBD6N9ef",
273
        "explorer_css_integrity": "sha384-vTFGj0krVqwFXLB7kq/VHR0/j2+cCT/B63rge2mULaqnib2OX7DVLUVksTlqvMab",
274
    }
275

276

277
def get_importmap() -> str:
9✔
278
    """Get the importmap for GraphiQL."""
279
    # Note that changing the versions here will break integrity checks! Regenerate: https://www.srihash.org/
280
    react = "https://esm.sh/react@19.1.0"
9✔
281
    react_jsx = "https://esm.sh/react@19.1.0/jsx-runtime"
9✔
282
    react_dom = "https://esm.sh/react-dom@19.1.0"
9✔
283
    react_dom_client = "https://esm.sh/react-dom@19.1.0/client"
9✔
284
    graphiql = "https://esm.sh/graphiql@5.0.3?standalone&external=react,react-dom,@graphiql/react,graphql"
9✔
285
    explorer = "https://esm.sh/@graphiql/plugin-explorer@5.0.0?standalone&external=react,@graphiql/react,graphql"
9✔
286
    graphiql_react = "https://esm.sh/@graphiql/react@0.35.4?standalone&external=react,react-dom,graphql"
9✔
287
    graphiql_toolkit = "https://esm.sh/@graphiql/toolkit@0.11.3?standalone&external=graphql"
9✔
288
    graphql = "https://esm.sh/graphql@16.11.0"
9✔
289
    json_worker = "https://esm.sh/monaco-editor@0.52.2/esm/vs/language/json/json.worker.js?worker"
9✔
290
    editor_worker = "https://esm.sh/monaco-editor@0.52.2/esm/vs/editor/editor.worker.js?worker"
9✔
291
    graphql_worker = "https://esm.sh/monaco-graphql@1.7.1/esm/graphql.worker.js?worker"
9✔
292

293
    importmap = {
9✔
294
        "imports": {
295
            "react": react,
296
            "react/jsx-runtime": react_jsx,
297
            "react-dom": react_dom,
298
            "react-dom/client": react_dom_client,
299
            "graphiql": graphiql,
300
            "@graphiql/plugin-explorer": explorer,
301
            "@graphiql/react": graphiql_react,
302
            "@graphiql/toolkit": graphiql_toolkit,
303
            "graphql": graphql,
304
            "monaco-editor/json-worker": json_worker,
305
            "monaco-editor/editor-worker": editor_worker,
306
            "monaco-graphql/graphql-worker": graphql_worker,
307
        },
308
        "integrity": {
309
            react: "sha384-C3ApUaeHIj1v0KX4cY/+K3hQZ/8HcAbbmkw1gBK8H5XN4LCEguY7+A3jga11SaHF",
310
            react_jsx: "sha384-ISrauaZAJlw0FRGhk9DBbU+2n4Bs1mrmh1kkJ63lTmkLTXYqpWTNFkGLPcK9C9BX",
311
            react_dom: "sha384-CKiqgCWLo5oVMbiCv36UR0pLRrzeRKhw1jFUpx0j/XdZOpZ43zOHhjf8yjLNuLEy",
312
            react_dom_client: "sha384-QH8CM8CiVIQ+RoTOjDp6ktXLkc0ix+qbx2mo7SSnwMeUQEoM4XJffjoSPY85X6VH",
313
            graphiql: "sha384-Vxuid6El2THg0+qYzVT1pHMOPQe1KZHNpxKaxqMqK4lbqokaJ0H+iKZR1zFhQzBN",
314
            explorer: "sha384-nrr4ZBQS9iyn0dn60BH3wtgG6YCSsQsuTPLgWDrUvvGvLvz0nE7LxnWWxEicmKHm",
315
            graphiql_react: "sha384-EnyV8FGnqfde43nPpmKz4yVI0VxlcwLVQe6P2r/cc24UcTmAzPU6SAabBonnSrT/",
316
            graphiql_toolkit: "sha384-ZsnupyYmzpNjF1Z/81zwi4nV352n4P7vm0JOFKiYnAwVGOf9twnEMnnxmxabMBXe",
317
            graphql: "sha384-uhRXaGfgCFqosYlwSLNd7XpDF9kcSUycv5yVbjjhH5OrE675kd0+MNIAAaSc+1Pi",
318
            json_worker: "sha384-8UXA1aePGFu/adc7cEQ8PPlVJityyzV0rDqM9Tjq1tiFFT0E7jIDQlOS4X431c+O",
319
            editor_worker: "sha384-lvRBk9hT6IKcVMAynOrBJUj/OCVkEaWBvzZdzvpPUqdrPW5bPsIBF7usVLLkQQxa",
320
            graphql_worker: "sha384-74lJ0Y2S6U0jkJAi5ijRRWnLiF0Fugr65EE1DVtJn/LHWmXJq9cVDFfC0eRjFkm1",
321
        },
322
    }
323
    return json.dumps(importmap, indent=2)
9✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc