Skip to content

Commit

Permalink
feat!: Update request.py in order to allow Streaming responses (#737)
Browse files Browse the repository at this point in the history
This was needed to build ChatGPT XBlocks.
  • Loading branch information
TheoBessel committed May 15, 2024
1 parent 7b17d4b commit 7d72ad0
Show file tree
Hide file tree
Showing 2 changed files with 50 additions and 8 deletions.
21 changes: 14 additions & 7 deletions xblock/django/request.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,24 @@
from lazy import lazy

import webob
from django.http import HttpResponse
from django.http import StreamingHttpResponse, HttpResponse
from webob.multidict import MultiDict, NestedMultiDict, NoVars


def webob_to_django_response(webob_response):
def webob_to_django_response(webob_response, streaming=False):
"""Returns a django response to the `webob_response`"""
django_response = HttpResponse(
webob_response.app_iter,
content_type=webob_response.content_type,
status=webob_response.status_code,
)
if streaming:
django_response = StreamingHttpResponse(
webob_response.app_iter,
content_type=webob_response.content_type,
status=webob_response.status_code,
)
else:
django_response = HttpResponse(
webob_response.app_iter,
content_type=webob_response.content_type,
status=webob_response.status_code,
)
for name, value in webob_response.headerlist:
django_response[name] = value
return django_response
Expand Down
37 changes: 36 additions & 1 deletion xblock/test/django/test_request.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ def test_post_already_read(self):
@pytest.mark.skipif(not HAS_DJANGO, reason='Django not available')
class TestDjangoWebobResponse(TestCase):
"""
Tests of the webob_to_django_response function
Tests of the webob_to_django_response function with streaming=False
"""
def _as_django(self, *args, **kwargs):
"""
Expand Down Expand Up @@ -82,3 +82,38 @@ def test_content_types(self):
self._as_django(content_type='text/html')['Content-Type'],
'text/html; charset=UTF-8'
)


@pytest.mark.skipif(not HAS_DJANGO, reason='Django not available')
class TestDjangoWebobResponseStreamed(TestCase):
"""
Tests of the webob_to_django_response function with streaming=True
"""
def _as_django(self, *args, **kwargs):
"""
Return a :class:`django.http.HttpResponse` created from a `webob.Response`
initialized with `*args` and `**kwargs`
"""
return webob_to_django_response(Response(*args, **kwargs), streaming=True)

def test_status_code(self):
self.assertEqual(self._as_django(status=200).status_code, 200)
self.assertEqual(self._as_django(status=404).status_code, 404)
self.assertEqual(self._as_django(status=500).status_code, 500)

def test_headers(self):
self.assertIn('X-Foo', self._as_django(headerlist=[('X-Foo', 'bar')]))
self.assertEqual(self._as_django(headerlist=[('X-Foo', 'bar')])['X-Foo'], 'bar')

def test_content_types(self):
# JSON content type (no charset should be returned)
self.assertEqual(
self._as_django(content_type='application/json')['Content-Type'],
'application/json'
)

# HTML content type (UTF-8 charset should be returned)
self.assertEqual(
self._as_django(content_type='text/html')['Content-Type'],
'text/html; charset=UTF-8'
)

0 comments on commit 7d72ad0

Please sign in to comment.