Feature allow to set the content-type of file uploads (#386)

Author: leszekhanusz

docs/usage/file_upload.rst

@@ -42,6 +42,25 @@ In order to upload a single file, you need to:
             query, variable_values=params, upload_files=True
         )
 
+Setting the content-type
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+If you need to set a specific Content-Type attribute to a file,
+you can set the :code:`content_type` attribute of the file like this:
+
+.. code-block:: python
+
+    with open("YOUR_FILE_PATH", "rb") as f:
+
+        # Setting the content-type to a pdf file for example
+        f.content_type = "application/pdf"
+
+        params = {"file": f}
+
+        result = client.execute(
+            query, variable_values=params, upload_files=True
+        )
+
 File list
 ---------
 

gql/transport/aiohttp.py

@@ -274,8 +274,11 @@ async def execute(
             data.add_field("map", file_map_str, content_type="application/json")
 
             # Add the extracted files as remaining fields
-            for k, v in file_streams.items():
-                data.add_field(k, v, filename=getattr(v, "name", k))
+            for k, f in file_streams.items():
+                name = getattr(f, "name", k)
+                content_type = getattr(f, "content_type", None)
+
+                data.add_field(k, f, filename=name, content_type=content_type)
 
             post_args: Dict[str, Any] = {"data": data}
 

gql/transport/httpx.py

@@ -103,9 +103,9 @@ def _prepare_file_uploads(self, variable_values, payload) -> Dict[str, Any]:
         # Prepare to send multipart-encoded data
         data: Dict[str, Any] = {}
         file_map: Dict[str, List[str]] = {}
-        file_streams: Dict[str, Tuple[str, Any]] = {}
+        file_streams: Dict[str, Tuple[str, ...]] = {}
 
-        for i, (path, val) in enumerate(files.items()):
+        for i, (path, f) in enumerate(files.items()):
             key = str(i)
 
             # Generate the file map
@@ -117,8 +117,13 @@ def _prepare_file_uploads(self, variable_values, payload) -> Dict[str, Any]:
             # Generate the file streams
             # Will generate something like
             # {"0": ("variables.file", <_io.BufferedReader ...>)}
-            filename = cast(str, getattr(val, "name", key))
-            file_streams[key] = (filename, val)
+            name = cast(str, getattr(f, "name", key))
+            content_type = getattr(f, "content_type", None)
+
+            if content_type is None:
+                file_streams[key] = (name, f)
+            else:
+                file_streams[key] = (name, f, content_type)
 
         # Add the payload to the operations field
         operations_str = self.json_serialize(payload)

gql/transport/requests.py

@@ -183,8 +183,14 @@ def execute(  # type: ignore
             fields = {"operations": operations_str, "map": file_map_str}
 
             # Add the extracted files as remaining fields
-            for k, v in file_streams.items():
-                fields[k] = (getattr(v, "name", k), v)
+            for k, f in file_streams.items():
+                name = getattr(f, "name", k)
+                content_type = getattr(f, "content_type", None)
+
+                if content_type is None:
+                    fields[k] = (name, f)
+                else:
+                    fields[k] = (name, f, content_type)
 
             # Prepare requests http to send multipart-encoded data
             data = MultipartEncoder(fields=fields)

tests/test_aiohttp.py

@@ -670,6 +670,74 @@ async def test_aiohttp_file_upload(event_loop, aiohttp_server):
             assert success
 
 
+async def single_upload_handler_with_content_type(request):
+
+    from aiohttp import web
+
+    reader = await request.multipart()
+
+    field_0 = await reader.next()
+    assert field_0.name == "operations"
+    field_0_text = await field_0.text()
+    assert field_0_text == file_upload_mutation_1_operations
+
+    field_1 = await reader.next()
+    assert field_1.name == "map"
+    field_1_text = await field_1.text()
+    assert field_1_text == file_upload_mutation_1_map
+
+    field_2 = await reader.next()
+    assert field_2.name == "0"
+    field_2_text = await field_2.text()
+    assert field_2_text == file_1_content
+
+    # Verifying the content_type
+    assert field_2.headers["Content-Type"] == "application/pdf"
+
+    field_3 = await reader.next()
+    assert field_3 is None
+
+    return web.Response(text=file_upload_server_answer, content_type="application/json")
+
+
+@pytest.mark.asyncio
+async def test_aiohttp_file_upload_with_content_type(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.aiohttp import AIOHTTPTransport
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler_with_content_type)
+    server = await aiohttp_server(app)
+
+    url = server.make_url("/")
+
+    transport = AIOHTTPTransport(url=url, timeout=10)
+
+    with TemporaryFile(file_1_content) as test_file:
+
+        async with Client(transport=transport) as session:
+
+            query = gql(file_upload_mutation_1)
+
+            file_path = test_file.filename
+
+            with open(file_path, "rb") as f:
+
+                # Setting the content_type
+                f.content_type = "application/pdf"
+
+                params = {"file": f, "other_var": 42}
+
+                # Execute query asynchronously
+                result = await session.execute(
+                    query, variable_values=params, upload_files=True
+                )
+
+            success = result["success"]
+
+            assert success
+
+
 @pytest.mark.asyncio
 async def test_aiohttp_file_upload_without_session(
     event_loop, aiohttp_server, run_sync_test

tests/test_httpx.py

@@ -477,6 +477,74 @@ def test_code():
     await run_sync_test(event_loop, server, test_code)
 
 
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_with_content_type(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def single_upload_handler(request):
+        from aiohttp import web
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_1_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_1_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        # Verifying the content_type
+        assert field_2.headers["Content-Type"] == "application/pdf"
+
+        field_3 = await reader.next()
+        assert field_3 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with TemporaryFile(file_1_content) as test_file:
+            with Client(transport=transport) as session:
+                query = gql(file_upload_mutation_1)
+
+                file_path = test_file.filename
+
+                with open(file_path, "rb") as f:
+
+                    # Setting the content_type
+                    f.content_type = "application/pdf"
+
+                    params = {"file": f, "other_var": 42}
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+    await run_sync_test(event_loop, server, test_code)
+
+
 @pytest.mark.aiohttp
 @pytest.mark.asyncio
 async def test_httpx_file_upload_additional_headers(

tests/test_requests.py

@@ -479,6 +479,74 @@ def test_code():
     await run_sync_test(event_loop, server, test_code)
 
 
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_requests_file_upload_with_content_type(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.requests import RequestsHTTPTransport
+
+    async def single_upload_handler(request):
+        from aiohttp import web
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_1_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_1_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        # Verifying the content_type
+        assert field_2.headers["Content-Type"] == "application/pdf"
+
+        field_3 = await reader.next()
+        assert field_3 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = server.make_url("/")
+
+    def test_code():
+        transport = RequestsHTTPTransport(url=url)
+
+        with TemporaryFile(file_1_content) as test_file:
+            with Client(transport=transport) as session:
+                query = gql(file_upload_mutation_1)
+
+                file_path = test_file.filename
+
+                with open(file_path, "rb") as f:
+
+                    # Setting the content_type
+                    f.content_type = "application/pdf"
+
+                    params = {"file": f, "other_var": 42}
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+    await run_sync_test(event_loop, server, test_code)
+
+
 @pytest.mark.aiohttp
 @pytest.mark.asyncio
 async def test_requests_file_upload_additional_headers(