Skip to content
Merged
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
12 changes: 12 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -527,6 +527,18 @@ result = await agent.run(
print(result.output)
```

## Uploading Files

`files.upload` takes a file path, raw bytes, or a binary file object. Files over 100 MB go through a presigned storage URL.

```python
file = metorial.files.upload("./report.pdf")

logo = metorial.files.upload("./logo.png", purpose="organization_image")

notes = metorial.files.upload(b"hello", file_name="notes.txt", title="Notes")
```

## Error Handling

```python
Expand Down
7 changes: 6 additions & 1 deletion src/metorial/_endpoint/endpoint_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,8 +47,12 @@ def _request(self, method: str, request: MetorialRequest, try_count: int = 0) ->
has_body = method in {"POST", "PUT", "PATCH"}
json_payload = None
files_payload = None
data_payload = None

if has_body and request.body is not None:
if has_body and request.files is not None:
files_payload = request.files
data_payload = request.data
elif has_body and request.body is not None:
# If body is a file-like object, send as multipart
if hasattr(request.body, "read"):
files_payload = request.body
Expand All @@ -67,6 +71,7 @@ def _request(self, method: str, request: MetorialRequest, try_count: int = 0) ->
headers=headers,
json=json_payload,
files=files_payload,
data=data_payload,
allow_redirects=True,
timeout=30,
)
Expand Down
9 changes: 8 additions & 1 deletion src/metorial/_endpoint/request.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,23 @@


class MetorialRequest:
"""Request configuration object for Metorial API calls."""
"""Request configuration object for Metorial API calls.

``files`` and ``data`` send a multipart body instead of JSON.
"""

def __init__(
self,
path: str | list[str],
host: str | None = None,
query: dict[str, Any] | None = None,
body: dict[str, Any] | None = None,
files: dict[str, Any] | None = None,
data: dict[str, str] | None = None,
):
self.path = path
self.host = host
self.query = query
self.body = body
self.files = files
self.data = data
217 changes: 217 additions & 0 deletions src/metorial/_file_upload.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,217 @@
from __future__ import annotations

import io
import mimetypes
import os
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from typing import IO, Any, Union

import requests

from metorial._endpoint import MetorialEndpointManager, MetorialRequest
from metorial._generated.magnetar.resources import (
DashboardInstanceFilesGetOutput,
mapDashboardInstanceFilesGetOutput,
)
from metorial.exceptions import MetorialSDKError

# Must stay in sync with maxDirectUploadSize in the files API.
MAX_DIRECT_UPLOAD_SIZE = 100 * 1024 * 1024

UPLOAD_TIMEOUT_SECONDS = 30

FileSource = Union[str, "os.PathLike[str]", bytes, IO[bytes]]


@dataclass
class _Source:
stream: IO[bytes]
name: str
content_type: str
size: int


def upload_file(
manager: MetorialEndpointManager,
file: FileSource,
*,
purpose: str,
title: str | None,
file_name: str | None,
content_type: str | None,
store_id: str | None,
store_path: str | None,
store_replace: bool,
instance_id: str | None,
direct_limit: int = MAX_DIRECT_UPLOAD_SIZE,
) -> DashboardInstanceFilesGetOutput:
if (store_id is None) != (store_path is None):
raise ValueError("store_id and store_path must be given together")
if store_replace and store_id is None:
raise ValueError("store_replace requires store_id and store_path")

fields: dict[str, Any] = {"purpose": purpose}
if title is not None:
fields["title"] = title
if instance_id is not None:
fields["instance_id"] = instance_id
if store_id is not None:
fields["store_id"] = store_id
fields["path"] = store_path
fields["store_replace"] = store_replace

with _open_source(file, file_name, content_type) as source:
if source.size <= direct_limit:
data = _upload_direct(manager, source, fields)
else:
data = _upload_with_url(manager, source, fields, instance_id)

return mapDashboardInstanceFilesGetOutput.from_dict(data)


def _upload_direct(
manager: MetorialEndpointManager, source: _Source, fields: dict[str, Any]
) -> Any:
form = {
key: ("true" if value else "false") if isinstance(value, bool) else str(value)
for key, value in fields.items()
}
# Read into memory so the 429 retry in the endpoint manager can resend the body.
content = source.stream.read()
request = MetorialRequest(
path=["files"],
files={"file": (source.name, content, source.content_type)},
data=form,
)
return manager._request("POST", request)


def _upload_with_url(
manager: MetorialEndpointManager,
source: _Source,
fields: dict[str, Any],
instance_id: str | None,
) -> Any:
session = manager._request(
"POST",
MetorialRequest(
path=["files"],
body={
"mode": "get_upload_url",
"file_name": source.name,
"file_size": source.size,
"file_type": source.content_type,
**fields,
},
),
)

upload = session.get("upload") if isinstance(session, dict) else None
if not isinstance(upload, dict) or not upload.get("url") or not session.get("id"):
raise MetorialSDKError(
{
"status": 0,
"code": "malformed_response",
"message": "The Metorial API did not return an upload URL.",
}
)

_put_to_storage(upload.get("method") or "PUT", upload["url"], source)

complete: dict[str, Any] = {"mode": "complete", "file_upload_id": session["id"]}
if instance_id is not None:
complete["instance_id"] = instance_id
return manager._request("POST", MetorialRequest(path=["files"], body=complete))


def _put_to_storage(method: str, url: str, source: _Source) -> None:
# The URL signature covers the content type, so it must match the declared
# file_type, and Metorial credentials must not be sent to the storage backend.
try:
response = requests.request(
method,
url,
data=source.stream,
headers={
"Content-Type": source.content_type,
"Content-Length": str(source.size),
},
timeout=UPLOAD_TIMEOUT_SECONDS,
)
except Exception as error:
raise MetorialSDKError(
{
"status": 0,
"code": "network_error",
"message": f"Failed to upload {source.name} to storage",
"error": str(error),
}
) from error

if not response.ok:
raise MetorialSDKError(
{
"status": response.status_code,
"code": "upload_failed",
"message": f"Failed to upload {source.name}: storage responded with {response.status_code}",
"body_snippet": (response.text or "")[:1000],
}
)


@contextmanager
def _open_source(
file: FileSource, file_name: str | None, content_type: str | None
) -> Iterator[_Source]:
if isinstance(file, (bytes, bytearray)):
if not file_name:
raise ValueError("file_name is required when uploading bytes")
yield _Source(
stream=io.BytesIO(file),
name=file_name,
content_type=content_type or _guess_content_type(file_name),
size=len(file),
)
return

if isinstance(file, (str, os.PathLike)):
path = os.path.abspath(os.fspath(file))
if not os.path.isfile(path):
raise ValueError(f"{os.fspath(file)} is not a file")

name = file_name or os.path.basename(path)
with open(path, "rb") as stream:
yield _Source(
stream=stream,
name=name,
content_type=content_type or _guess_content_type(name),
size=os.fstat(stream.fileno()).st_size,
)
return

name = file_name or os.path.basename(str(getattr(file, "name", "") or ""))
if not name:
raise ValueError("file_name is required when uploading a stream without a name")

if file.seekable():
start = file.tell()
size = file.seek(0, io.SEEK_END) - start
file.seek(start)
readable: IO[bytes] = file
else:
content = file.read()
readable = io.BytesIO(content)
size = len(content)

yield _Source(
stream=readable,
name=name,
content_type=content_type or _guess_content_type(name),
size=size,
)


def _guess_content_type(name: str) -> str:
return mimetypes.guess_type(name)[0] or "application/octet-stream"
35 changes: 35 additions & 0 deletions src/metorial/_magnetar_sdk.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import httpx

from metorial._endpoint import MetorialEndpointManager
from metorial._file_upload import FileSource, upload_file
from metorial._generated.magnetar.endpoints.callback_events import (
MetorialCallbackEventsEndpoint,
)
Expand Down Expand Up @@ -336,6 +337,7 @@
from metorial._generated.magnetar.endpoints.tool_calls import (
MetorialToolCallsEndpoint,
)
from metorial._generated.magnetar.resources import DashboardInstanceFilesGetOutput
from metorial._sdk_builder import MetorialSDKBuilder
from metorial._sdk_shared import SDKConfig, get_api_host, get_headers
from metorial.sandbox import (
Expand Down Expand Up @@ -717,6 +719,39 @@ class MagnetarFilesGroup(MetorialFilesEndpoint):
def __init__(self, manager: MetorialEndpointManager) -> None:
super().__init__(manager)
self.links = MetorialFilesLinksEndpoint(manager)
self._manager = manager

def upload(
self,
file: FileSource,
*,
purpose: str = "generic",
title: str | None = None,
file_name: str | None = None,
content_type: str | None = None,
store_id: str | None = None,
store_path: str | None = None,
store_replace: bool = False,
instance_id: str | None = None,
) -> DashboardInstanceFilesGetOutput:
"""Upload a file and return the created file.

``file`` is a file path, raw bytes, or a binary file object. Bytes and
unnamed streams need ``file_name``. Files over 100 MB are uploaded
through a presigned storage URL.
"""
return upload_file(
self._manager,
file,
purpose=purpose,
title=title,
file_name=file_name,
content_type=content_type,
store_id=store_id,
store_path=store_path,
store_replace=store_replace,
instance_id=instance_id,
)


class MagnetarSkillsGroup(MetorialSkillsEndpoint):
Expand Down
Loading
Loading