Simple support for range headers

starlette/responses.py

@@ -20,6 +20,7 @@
 try:
     import aiofiles
     from aiofiles.os import stat as aio_stat
+    from aiofiles.threadpool import open as aio_open
 except ImportError:  # pragma: nocover
     aiofiles = None  # type: ignore
     aio_stat = None  # type: ignore
@@ -243,6 +244,7 @@ def __init__(
         filename: str = None,
         stat_result: os.stat_result = None,
         method: str = None,
+        offset: int = 0,
     ) -> None:
         assert aiofiles is not None, "'aiofiles' must be installed to use FileResponse"
         self.path = path
@@ -266,6 +268,7 @@ def __init__(
         self.stat_result = stat_result
         if stat_result is not None:
             self.set_stat_headers(stat_result)
+        self.offset = offset
 
     def set_stat_headers(self, stat_result: os.stat_result) -> None:
         content_length = str(stat_result.st_size)
@@ -301,7 +304,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             # Tentatively ignoring type checking failure to work around the wrong type
             # definitions for aiofile that come with typeshed. See
             # https://github.com/python/typeshed/pull/4650
-            async with aiofiles.open(self.path, mode="rb") as file:  # type: ignore
+            async with aio_open(self.path, mode="rb") as file:  # type: ignore
+                await file.seek(self.offset)
                 more_body = True
                 while more_body:
                     chunk = await file.read(self.chunk_size)

tests/test_responses.py

@@ -244,6 +244,19 @@ def test_file_response_with_chinese_filename(tmpdir):
     assert response.headers["content-disposition"] == expected_disposition
 
 
+def test_file_response_with_offset(tmpdir):
+    content = b"000 111 222 333 444 555"
+    filename = "offset"
+    path = os.path.join(tmpdir, filename)
+    with open(path, "wb") as f:
+        f.write(content)
+    for offset in range(0, 24, 4):  # skip blocks of 4
+        app = FileResponse(path=path, offset=offset)
+        client = TestClient(app)
+        response = client.get("/")
+        assert response.content == content[offset:]
+
+
 def test_set_cookie():
     async def app(scope, receive, send):
         response = Response("Hello, world!", media_type="text/plain")