Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 65 additions & 1 deletion docs/advanced.md
Original file line number Diff line number Diff line change
Expand Up @@ -66,4 +66,68 @@ nb = pynetbox.api(
token='d6f4e314a5b5fefd164995169f28ae32d987704f'
)
nb.http_session = session
```
```

# 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)`.
78 changes: 72 additions & 6 deletions pynetbox/core/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,54 @@
"""

import concurrent.futures as cf
import io
import os
import json

from packaging import version


def _is_file_like(obj):
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.

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)
Expand Down Expand Up @@ -257,10 +300,23 @@ 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
# 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)

# 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)
Expand All @@ -272,9 +328,19 @@ 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)
Expand Down
Loading