Better async file handling

CHANGELOG.rst

@@ -11,3 +11,8 @@ Releases
 ------------------
 
 * Corrected an error in the README
+
+0.0.3 (2025-10-28)
+------------------
+
+* Improved asynchronous file handling in middleware

Makefile

@@ -1,6 +1,6 @@
 .PHONY: init
 init:
-	python3 -m virtualenv .venv
+	python3 -m venv .venv
 	source .venv/bin/activate
 	pip install -r requirements.txt
 
@@ -17,10 +17,6 @@ ruff:
 	ruff check .
 	ruff format --check .
 
-.PHONY: mypy
-mypy:
-	mypy .
-
 .PHONY: coverage
 coverage:
 	pytest --verbose --cov-report term --cov-report html --cov=django_sam tests

django_sam/__init__.py

@@ -1 +1 @@
-__version__ = '0.0.2'
+__version__ = '0.0.3'

django_sam/middleware.py

@@ -1,9 +1,11 @@
 import logging
+import mimetypes
 import os
 
+import aiofiles
 from asgiref.sync import iscoroutinefunction
 from django.conf import settings
-from django.http import FileResponse
+from django.http import FileResponse, StreamingHttpResponse
 from django.utils.decorators import sync_and_async_middleware
 
 logger = logging.getLogger(__name__)
@@ -38,7 +40,24 @@ def static_admin_middleware(get_response):
 
         async def middleware(request):
             if files.get(request.path):
-                return FileResponse(open(files[request.path], 'rb'))
+                file_path = files[request.path]
+
+                async def file_iterator():
+                    async with aiofiles.open(file_path, 'rb') as f:
+                        while chunk := await f.read(8192):
+                            yield chunk
+
+                content_type, _ = mimetypes.guess_type(file_path)
+                if content_type is None:
+                    content_type = 'application/octet-stream'
+
+                response = StreamingHttpResponse(
+                    file_iterator(), content_type=content_type
+                )
+
+                filename = os.path.basename(file_path)
+                response['Content-Disposition'] = f'inline; filename="{filename}"'
+                return response
 
             response = await get_response(request)
             return response

requirements.txt

@@ -1,5 +1,6 @@
 pytest>=8.0.0,<9.0.0
 pytest-cov>=3.0.0,<5.0.0
 pytest-mock>=3.13.0,<4.0.0
+aiofiles
 django>=3.2.0,<6.0.0
 setuptools

setup.py

@@ -34,17 +34,16 @@ def get_version():
     include_package_data=True,
     classifiers=[
         'Natural Language :: English',
-        'License :: OSI Approved :: MIT License',
         'Programming Language :: Python',
         'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.8',
         'Programming Language :: Python :: 3.9',
         'Programming Language :: Python :: 3.10',
         'Programming Language :: Python :: 3.11',
         'Programming Language :: Python :: 3.12',
     ],
-    python_requires='>=3.8',
+    python_requires='>=3.9',
     install_requires=[
+        'aiofiles',
         'django>=3.2.0',
     ],
 )

tests/test_middleware.py

@@ -1,4 +1,11 @@
-from django_sam.middleware import gather_files
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+from django.conf import settings
+from django.http import FileResponse, StreamingHttpResponse
+from django.test import RequestFactory
+
+from django_sam.middleware import gather_files, static_admin_middleware
 
 
 def test_gather_files_dict_with_valid_path(mocker):
@@ -30,3 +37,142 @@ def test_gather_files_logs_error_with_invalid_path(mocker):
     mock_logger.error.assert_called_once_with(
         "Static root directory '/srv/service/static/' does not exist."
     )
+
+
+class TestStaticAdminMiddleware:
+    def setup_method(self):
+        if not settings.configured:
+            settings.configure(
+                STATIC_ROOT='/test/static',
+                STATIC_URL='/static/',
+            )
+        self.factory = RequestFactory()
+        self.mock_files = {'/static/admin/css/admin.css': '/srv/service/static/admin/css/admin.css'}
+
+    def test_middleware_initialization_calls_gather_files(self, mocker):
+        mock_gather_files = mocker.patch('django_sam.middleware.gather_files', return_value={})
+
+        def get_response(request):
+            return MagicMock()
+
+        static_admin_middleware(get_response)
+        mock_gather_files.assert_called_once_with('/test/static', '/static/')
+
+    def test_sync_middleware_serves_static_file(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        mock_file = MagicMock()
+        mocker.patch('builtins.open', return_value=mock_file)
+
+        def get_response(request):
+            return MagicMock()
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/static/admin/css/admin.css')
+        response = middleware(request)
+
+        assert isinstance(response, FileResponse)
+        assert response.file_to_stream == mock_file
+
+    def test_sync_middleware_handles_non_existent_file(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        expected_response = MagicMock()
+        def get_response(request):
+            return expected_response
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/static/admin/css/nonexistent.css')
+        response = middleware(request)
+
+        assert response == expected_response
+
+    def test_sync_middleware_passes_through_non_static_requests(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        expected_response = MagicMock()
+        def get_response(request):
+            return expected_response
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/admin/')
+        response = middleware(request)
+
+        assert response == expected_response
+
+    def test_async_middleware_serves_static_file(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        mock_file = AsyncMock()
+        mock_file.read = AsyncMock(side_effect=[b'chunk1', b'chunk2', b''])
+        mock_aiofiles = mocker.patch('django_sam.middleware.aiofiles')
+        mock_aiofiles.open.return_value.__aenter__.return_value = mock_file
+
+        async def get_response(request):
+            return MagicMock()
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/static/admin/css/admin.css')
+        async def run_test():
+            response = await middleware(request)
+            return response
+        response = asyncio.run(run_test())
+
+        assert isinstance(response, StreamingHttpResponse)
+        assert response['Content-Type'] == 'text/css'
+        assert response['Content-Disposition'] == 'inline; filename="admin.css"'
+
+    def test_async_middleware_handles_nonexistent_file(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        expected_response = MagicMock()
+        async def get_response(request):
+            return expected_response
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/static/admin/css/nonexistent.css')
+        async def run_test():
+            response = await middleware(request)
+            return response
+        response = asyncio.run(run_test())
+
+        assert response == expected_response
+
+    def test_async_middleware_handles_unknown_mimetype(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        mock_file = AsyncMock()
+        mock_file.read = AsyncMock(side_effect=[b'content', b''])
+        mock_aiofiles = mocker.patch('django_sam.middleware.aiofiles')
+        mock_aiofiles.open.return_value.__aenter__.return_value = mock_file
+
+        mocker.patch('django_sam.middleware.mimetypes.guess_type', return_value=(None, None))
+
+        async def get_response(request):
+            return MagicMock()
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/static/admin/css/admin.css')
+        async def run_test():
+            response = await middleware(request)
+            return response
+        response = asyncio.run(run_test())
+
+        assert isinstance(response, StreamingHttpResponse)
+        assert response['Content-Type'] == 'application/octet-stream'
+
+    def test_async_middleware_passes_through_non_static_requests(self, mocker):
+        mocker.patch('django_sam.middleware.gather_files', return_value=self.mock_files)
+
+        expected_response = MagicMock()
+        async def get_response(request):
+            return expected_response
+
+        middleware = static_admin_middleware(get_response)
+        request = self.factory.get('/admin/')
+        async def run_test():
+            response = await middleware(request)
+            return response
+        response = asyncio.run(run_test())
+
+        assert response == expected_response