From 02e6b7686090f818c8f67680a00417fbc42be7b6 Mon Sep 17 00:00:00 2001 From: Xinhao Luo Date: Thu, 27 Nov 2025 02:37:15 -0800 Subject: [PATCH 1/4] feat(image): add image upload support --- docs/advanced.md | 66 ++++++++++- pynetbox/core/query.py | 60 +++++++++- tests/unit/test_file_upload.py | 193 +++++++++++++++++++++++++++++++++ 3 files changed, 314 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_file_upload.py diff --git a/docs/advanced.md b/docs/advanced.md index d3f0cc51..2061f86d 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -66,4 +66,68 @@ nb = pynetbox.api( token='d6f4e314a5b5fefd164995169f28ae32d987704f' ) nb.http_session = session -``` \ No newline at end of file +``` + +# File Uploads (Image Attachments) + +Pynetbox supports file uploads for endpoints that accept them, such as image attachments. When you pass a file-like object (anything with a `.read()` method) to `create()`, pynetbox automatically detects it and uses multipart/form-data encoding instead of JSON. + +## Creating an Image Attachment + +```python +import pynetbox + +nb = pynetbox.api( + 'http://localhost:8000', + token='d6f4e314a5b5fefd164995169f28ae32d987704f' +) + +# Attach an image to a device +with open('/path/to/image.png', 'rb') as f: + attachment = nb.extras.image_attachments.create( + object_type='dcim.device', + object_id=1, + image=f, + name='rack-photo.png' + ) +``` + +## Using io.BytesIO + +You can also use in-memory file objects: + +```python +import io +import pynetbox + +nb = pynetbox.api( + 'http://localhost:8000', + token='d6f4e314a5b5fefd164995169f28ae32d987704f' +) + +# Create image from bytes +image_data = b'...' # Your image bytes +file_obj = io.BytesIO(image_data) +file_obj.name = 'generated-image.png' # Optional: set filename + +attachment = nb.extras.image_attachments.create( + object_type='dcim.device', + object_id=1, + image=file_obj +) +``` + +## Custom Filename and Content-Type + +For more control, pass a tuple instead of a file object: + +```python +with open('/path/to/image.png', 'rb') as f: + attachment = nb.extras.image_attachments.create( + object_type='dcim.device', + object_id=1, + image=('custom-name.png', f, 'image/png') + ) +``` + +The tuple format is `(filename, file_object)` or `(filename, file_object, content_type)`. \ No newline at end of file diff --git a/pynetbox/core/query.py b/pynetbox/core/query.py index 278e2d80..64b26291 100644 --- a/pynetbox/core/query.py +++ b/pynetbox/core/query.py @@ -15,11 +15,49 @@ """ import concurrent.futures as cf +import os import json from packaging import version +def _is_file_like(obj): + """Check if an object is file-like (has a read method and is not a string/bytes).""" + return hasattr(obj, "read") and not isinstance(obj, (str, bytes)) + + +def _extract_files(data): + """Extract file-like objects from data dict. + + Returns a tuple of (clean_data, files) where clean_data has file objects + removed and files is a dict suitable for requests' files parameter. + """ + if not isinstance(data, dict): + return data, None + + files = {} + clean_data = {} + + for key, value in data.items(): + if _is_file_like(value): + # Format: (filename, file_obj, content_type) + # Try to get filename from file object, fallback to key + filename = getattr(value, "name", None) + if filename: + # Extract just the filename, not the full path + filename = os.path.basename(filename) + else: + filename = key + files[key] = (filename, value) + elif isinstance(value, tuple) and len(value) >= 2 and _is_file_like(value[1]): + # Already in (filename, file_obj) or (filename, file_obj, content_type) format + files[key] = value + else: + clean_data[key] = value + + return clean_data, files if files else None + + def calc_pages(limit, count): """Calculate number of pages required for full results set.""" return int(count / limit) + (limit % count > 0) @@ -257,7 +295,15 @@ def normalize_url(self, url): return url def _make_call(self, verb="get", url_override=None, add_params=None, data=None): - if verb in ("post", "put") or verb == "delete" and data: + # Extract any file-like objects from data + files = None + if data is not None and verb in ("post", "put", "patch"): + data, files = _extract_files(data) + + if files: + # For multipart/form-data, don't set Content-Type (requests handles it) + headers = {"accept": "application/json"} + elif verb in ("post", "put") or verb == "delete" and data: headers = {"Content-Type": "application/json"} else: headers = {"accept": "application/json"} @@ -272,9 +318,15 @@ def _make_call(self, verb="get", url_override=None, add_params=None, data=None): if add_params: params.update(add_params) - req = getattr(self.http_session, verb)( - url_override or self.url, headers=headers, params=params, json=data - ) + if files: + # Use multipart/form-data for file uploads + req = getattr(self.http_session, verb)( + url_override or self.url, headers=headers, params=params, data=data, files=files + ) + else: + req = getattr(self.http_session, verb)( + url_override or self.url, headers=headers, params=params, json=data + ) if req.status_code == 409 and verb == "post": raise AllocationError(req) diff --git a/tests/unit/test_file_upload.py b/tests/unit/test_file_upload.py new file mode 100644 index 00000000..fc35e2b6 --- /dev/null +++ b/tests/unit/test_file_upload.py @@ -0,0 +1,193 @@ +"""Tests for file upload/multipart support.""" + +import io +import unittest +from unittest.mock import Mock, patch + +from pynetbox.core.query import Request, _extract_files, _is_file_like +from pynetbox.core.endpoint import Endpoint + +class TestIsFileLike(unittest.TestCase): + """Tests for _is_file_like helper function.""" + + def test_file_object(self): + """File objects should be detected as file-like.""" + f = io.BytesIO(b"test content") + self.assertTrue(_is_file_like(f)) + + def test_string_io(self): + """StringIO objects should be detected as file-like.""" + f = io.StringIO("test content") + self.assertTrue(_is_file_like(f)) + + def test_string(self): + """Strings should not be detected as file-like.""" + self.assertFalse(_is_file_like("test")) + + def test_bytes(self): + """Bytes should not be detected as file-like.""" + self.assertFalse(_is_file_like(b"test")) + + def test_dict(self): + """Dicts should not be detected as file-like.""" + self.assertFalse(_is_file_like({"key": "value"})) + + def test_none(self): + """None should not be detected as file-like.""" + self.assertFalse(_is_file_like(None)) + + +class TestExtractFiles(unittest.TestCase): + """Tests for _extract_files helper function.""" + + def test_no_files(self): + """Data without files should return unchanged.""" + data = {"name": "test", "device": 1} + clean_data, files = _extract_files(data) + self.assertEqual(clean_data, {"name": "test", "device": 1}) + self.assertIsNone(files) + + def test_extract_file_object(self): + """File objects should be extracted from data.""" + file_obj = io.BytesIO(b"test content") + data = {"name": "test", "image": file_obj} + clean_data, files = _extract_files(data) + + self.assertEqual(clean_data, {"name": "test"}) + self.assertIn("image", files) + self.assertEqual(files["image"][0], "image") # filename defaults to key + self.assertEqual(files["image"][1], file_obj) + + def test_extract_file_with_name_attribute(self): + """File objects with name attribute should use that as filename.""" + file_obj = io.BytesIO(b"test content") + file_obj.name = "/path/to/image.png" + data = {"name": "test", "image": file_obj} + clean_data, files = _extract_files(data) + + self.assertEqual(files["image"][0], "image.png") # basename extracted + + def test_tuple_format(self): + """Files passed as tuples should be preserved.""" + file_obj = io.BytesIO(b"test content") + data = {"name": "test", "image": ("custom_name.png", file_obj, "image/png")} + clean_data, files = _extract_files(data) + + self.assertEqual(clean_data, {"name": "test"}) + self.assertEqual(files["image"], ("custom_name.png", file_obj, "image/png")) + + def test_non_dict_data(self): + """Non-dict data should be returned unchanged.""" + data = [{"id": 1}, {"id": 2}] + result_data, files = _extract_files(data) + self.assertEqual(result_data, data) + self.assertIsNone(files) + + +class TestRequestWithFiles(unittest.TestCase): + """Tests for Request._make_call with file uploads.""" + + def test_post_with_files_uses_multipart(self): + """POST with files should use multipart/form-data.""" + mock_session = Mock() + mock_session.post.return_value.ok = True + mock_session.post.return_value.status_code = 201 + mock_session.post.return_value.json.return_value = {"id": 1, "name": "test"} + + file_obj = io.BytesIO(b"test content") + req = Request( + base="http://localhost:8000/api/extras/image-attachments", + http_session=mock_session, + token="testtoken", + ) + req._make_call( + verb="post", + data={"object_type": "dcim.device", "object_id": 1, "image": file_obj}, + ) + + # Verify multipart was used (data= instead of json=) + mock_session.post.assert_called_once() + call_kwargs = mock_session.post.call_args + self.assertIn("data", call_kwargs.kwargs) + self.assertIn("files", call_kwargs.kwargs) + self.assertNotIn("json", call_kwargs.kwargs) + + # Verify Content-Type was not set (requests handles it for multipart) + headers = call_kwargs.kwargs["headers"] + self.assertNotIn("Content-Type", headers) + self.assertEqual(headers["accept"], "application/json") + self.assertEqual(headers["authorization"], "Token testtoken") + + def test_post_without_files_uses_json(self): + """POST without files should use JSON.""" + mock_session = Mock() + mock_session.post.return_value.ok = True + mock_session.post.return_value.status_code = 201 + mock_session.post.return_value.json.return_value = {"id": 1, "name": "test"} + + req = Request( + base="http://localhost:8000/api/dcim/devices", + http_session=mock_session, + token="testtoken", + ) + req._make_call(verb="post", data={"name": "test-device", "site": 1}) + + mock_session.post.assert_called_once() + call_kwargs = mock_session.post.call_args + self.assertIn("json", call_kwargs.kwargs) + self.assertNotIn("files", call_kwargs.kwargs) + self.assertEqual(call_kwargs.kwargs["headers"]["Content-Type"], "application/json") + + def test_patch_with_files_uses_multipart(self): + """PATCH with files should use multipart/form-data.""" + mock_session = Mock() + mock_session.patch.return_value.ok = True + mock_session.patch.return_value.status_code = 200 + mock_session.patch.return_value.json.return_value = {"id": 1, "name": "test"} + + file_obj = io.BytesIO(b"new content") + req = Request( + base="http://localhost:8000/api/extras/image-attachments", + http_session=mock_session, + token="testtoken", + key=1, + ) + req._make_call(verb="patch", data={"image": file_obj}) + + mock_session.patch.assert_called_once() + call_kwargs = mock_session.patch.call_args + self.assertIn("data", call_kwargs.kwargs) + self.assertIn("files", call_kwargs.kwargs) + + +class TestEndpointCreateWithFiles(unittest.TestCase): + """Tests for Endpoint.create() with file uploads.""" + + def test_create_image_attachment(self): + """Creating image attachment should work with file objects.""" + with patch("pynetbox.core.query.Request._make_call") as mock_call: + + mock_call.return_value = { + "id": 1, + "object_type": "dcim.device", + "object_id": 1, + "image": "/media/image-attachments/test.png", + "name": "test.png", + } + + api = Mock(base_url="http://localhost:8000/api") + app = Mock(name="extras") + endpoint = Endpoint(api, app, "image_attachments") + + file_obj = io.BytesIO(b"fake image content") + endpoint.create( + object_type="dcim.device", object_id=1, image=file_obj, name="test.png" + ) + + mock_call.assert_called_once() + call_kwargs = mock_call.call_args + self.assertEqual(call_kwargs.kwargs["verb"], "post") + data = call_kwargs.kwargs["data"] + self.assertEqual(data["object_type"], "dcim.device") + self.assertEqual(data["object_id"], 1) + self.assertEqual(data["image"], file_obj) \ No newline at end of file From f8a2ff8f3b791979f93b2f7a6c307ec70e13f19b Mon Sep 17 00:00:00 2001 From: Xinhao Luo Date: Tue, 2 Dec 2025 11:39:10 -0800 Subject: [PATCH 2/4] fix(query): set Content-Type header for PATCH requests --- pynetbox/core/query.py | 17 ++++++++++------- tests/unit/test_file_upload.py | 21 +++++++++++++++++++++ 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/pynetbox/core/query.py b/pynetbox/core/query.py index 64b26291..7b140d12 100644 --- a/pynetbox/core/query.py +++ b/pynetbox/core/query.py @@ -297,16 +297,19 @@ def normalize_url(self, url): def _make_call(self, verb="get", url_override=None, add_params=None, data=None): # Extract any file-like objects from data files = None - if data is not None and verb in ("post", "put", "patch"): + # Verbs that support request bodies with file uploads + body_verbs = ("post", "put", "patch") + headers = {"accept": "application/json"} + + # Extract files from data for applicable verbs + if data is not None and verb in body_verbs: data, files = _extract_files(data) - if files: - # For multipart/form-data, don't set Content-Type (requests handles it) - headers = {"accept": "application/json"} - elif verb in ("post", "put") or verb == "delete" and data: + # Set headers based on request type + should_be_json_body = not files and (verb in body_verbs or (verb == "delete" and data)) + + if should_be_json_body: headers = {"Content-Type": "application/json"} - else: - headers = {"accept": "application/json"} if self.token: headers["authorization"] = "Token {}".format(self.token) diff --git a/tests/unit/test_file_upload.py b/tests/unit/test_file_upload.py index fc35e2b6..4d85fd7d 100644 --- a/tests/unit/test_file_upload.py +++ b/tests/unit/test_file_upload.py @@ -159,6 +159,27 @@ def test_patch_with_files_uses_multipart(self): self.assertIn("data", call_kwargs.kwargs) self.assertIn("files", call_kwargs.kwargs) + def test_patch_without_files_uses_json(self): + """PATCH without files should use JSON and set Content-Type.""" + mock_session = Mock() + mock_session.patch.return_value.ok = True + mock_session.patch.return_value.status_code = 200 + mock_session.patch.return_value.json.return_value = {"id": 1, "name": "updated"} + + req = Request( + base="http://localhost:8000/api/dcim/devices", + http_session=mock_session, + token="testtoken", + key=1, + ) + req._make_call(verb="patch", data={"name": "updated-device"}) + + mock_session.patch.assert_called_once() + call_kwargs = mock_session.patch.call_args + self.assertIn("json", call_kwargs.kwargs) + self.assertNotIn("files", call_kwargs.kwargs) + self.assertEqual(call_kwargs.kwargs["headers"]["Content-Type"], "application/json") + class TestEndpointCreateWithFiles(unittest.TestCase): """Tests for Endpoint.create() with file uploads.""" From 6246763dc0c13c36a34f38590c275d5167b3793b Mon Sep 17 00:00:00 2001 From: Xinhao Luo Date: Tue, 2 Dec 2025 11:51:24 -0800 Subject: [PATCH 3/4] addressing comments --- pynetbox/core/query.py | 10 +++++++--- tests/unit/test_file_upload.py | 7 +++++++ 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/pynetbox/core/query.py b/pynetbox/core/query.py index 7b140d12..90cbb6aa 100644 --- a/pynetbox/core/query.py +++ b/pynetbox/core/query.py @@ -15,6 +15,7 @@ """ import concurrent.futures as cf +import io import os import json @@ -22,9 +23,12 @@ def _is_file_like(obj): - """Check if an object is file-like (has a read method and is not a string/bytes).""" - return hasattr(obj, "read") and not isinstance(obj, (str, bytes)) - + if isinstance(obj, (str, bytes)): + return False + # Check if it's a standard library IO object OR has a callable read method + return isinstance(obj, io.IOBase) or ( + hasattr(obj, "read") and callable(getattr(obj, "read")) + ) def _extract_files(data): """Extract file-like objects from data dict. diff --git a/tests/unit/test_file_upload.py b/tests/unit/test_file_upload.py index 4d85fd7d..6bb105d4 100644 --- a/tests/unit/test_file_upload.py +++ b/tests/unit/test_file_upload.py @@ -36,6 +36,13 @@ def test_none(self): """None should not be detected as file-like.""" self.assertFalse(_is_file_like(None)) + def test_non_callable_read_attribute(self): + """Objects with non-callable read attribute should not be file-like.""" + class FakeFile: + read = "not a method" + + self.assertFalse(_is_file_like(FakeFile())) + class TestExtractFiles(unittest.TestCase): """Tests for _extract_files helper function.""" From e82167721f75b1780b4bb84705585bffeecdc33d Mon Sep 17 00:00:00 2001 From: Xinhao Luo Date: Fri, 5 Dec 2025 01:25:07 -0800 Subject: [PATCH 4/4] addressing lint --- pynetbox/core/query.py | 11 +++++++++-- tests/unit/test_file_upload.py | 12 +++++++++--- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/pynetbox/core/query.py b/pynetbox/core/query.py index 90cbb6aa..2b4f4d0d 100644 --- a/pynetbox/core/query.py +++ b/pynetbox/core/query.py @@ -30,6 +30,7 @@ def _is_file_like(obj): hasattr(obj, "read") and callable(getattr(obj, "read")) ) + def _extract_files(data): """Extract file-like objects from data dict. @@ -310,7 +311,9 @@ def _make_call(self, verb="get", url_override=None, add_params=None, data=None): data, files = _extract_files(data) # Set headers based on request type - should_be_json_body = not files and (verb in body_verbs or (verb == "delete" and data)) + should_be_json_body = not files and ( + verb in body_verbs or (verb == "delete" and data) + ) if should_be_json_body: headers = {"Content-Type": "application/json"} @@ -328,7 +331,11 @@ def _make_call(self, verb="get", url_override=None, add_params=None, data=None): if files: # Use multipart/form-data for file uploads req = getattr(self.http_session, verb)( - url_override or self.url, headers=headers, params=params, data=data, files=files + url_override or self.url, + headers=headers, + params=params, + data=data, + files=files, ) else: req = getattr(self.http_session, verb)( diff --git a/tests/unit/test_file_upload.py b/tests/unit/test_file_upload.py index 6bb105d4..7a7f9f12 100644 --- a/tests/unit/test_file_upload.py +++ b/tests/unit/test_file_upload.py @@ -7,6 +7,7 @@ from pynetbox.core.query import Request, _extract_files, _is_file_like from pynetbox.core.endpoint import Endpoint + class TestIsFileLike(unittest.TestCase): """Tests for _is_file_like helper function.""" @@ -38,6 +39,7 @@ def test_none(self): def test_non_callable_read_attribute(self): """Objects with non-callable read attribute should not be file-like.""" + class FakeFile: read = "not a method" @@ -143,7 +145,9 @@ def test_post_without_files_uses_json(self): call_kwargs = mock_session.post.call_args self.assertIn("json", call_kwargs.kwargs) self.assertNotIn("files", call_kwargs.kwargs) - self.assertEqual(call_kwargs.kwargs["headers"]["Content-Type"], "application/json") + self.assertEqual( + call_kwargs.kwargs["headers"]["Content-Type"], "application/json" + ) def test_patch_with_files_uses_multipart(self): """PATCH with files should use multipart/form-data.""" @@ -185,7 +189,9 @@ def test_patch_without_files_uses_json(self): call_kwargs = mock_session.patch.call_args self.assertIn("json", call_kwargs.kwargs) self.assertNotIn("files", call_kwargs.kwargs) - self.assertEqual(call_kwargs.kwargs["headers"]["Content-Type"], "application/json") + self.assertEqual( + call_kwargs.kwargs["headers"]["Content-Type"], "application/json" + ) class TestEndpointCreateWithFiles(unittest.TestCase): @@ -218,4 +224,4 @@ def test_create_image_attachment(self): data = call_kwargs.kwargs["data"] self.assertEqual(data["object_type"], "dcim.device") self.assertEqual(data["object_id"], 1) - self.assertEqual(data["image"], file_obj) \ No newline at end of file + self.assertEqual(data["image"], file_obj)