API Reference
This document provides comprehensive documentation for the high-level Nexus API classes and methods.
Async is Default: The primary API is async-first (
miura.api). The sync API (miura) is a compatibility layer that wraps the async API.
API Layers
The Miura API is organized into clear layers:
- Primary API (
miura.api): Async-first high-level API - Recommended for all new code - Sync API (
miura): Synchronous wrappers for compatibility - Internal: Implementation details (marked with
_prefix) - Do not import directly
Note: Low-level clients (NexusClient) are internal implementation details and should not be imported directly. Use the high-level API (AsyncNexus or Nexus) instead.
Navigation Styles
The API supports two navigation styles, both equally supported:
- Object-Oriented Navigation: Navigate through objects (Project → Collection → Item)
- Path-Based Navigation: Navigate using hierarchical paths (e.g.,
"project/collection/item")
See the Projects and Collections tutorial for detailed guidance on when to use each style.
Async API (miura.api)
The async API is the primary, canonical interface. Use it for all new code.
from miura.api import AsyncNexus, AsyncProject, AsyncCollection
Note: Imports work without side effects (no network calls on import).
AsyncNexus
Main async client class for interacting with the Nexus platform.
Constructor
async with AsyncNexus(
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
verify_ssl: bool = True,
) as nexus:
# Use nexus here
Parameters:
retry_policy(OptionalRetryPolicy): Retry policy configurationtimeouts(OptionalTimeouts): Timeout configurationhooks(OptionalApiHooks): Callback hooks for observabilityverify_ssl(bool): Whether to verify SSL certificates (default: True)
Methods
async def list_projects(
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> list[ProjectInfo]
List all projects accessible to the current user.
Returns: List of ProjectInfo objects (dataclass instances with attribute access)
Example:
from miura.logging import get_logger
logger = get_logger(__name__)
projects = await nexus.list_projects()
for project in projects:
logger.info(f"Project: {project.name}")
<|tool▁calls▁begin|><|tool▁call▁begin|> grep
get_project(project_name: str)
async def get_project(
project_name: str,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncProject
Get a project by name.
Parameters:
project_name(str): Name of the project
Returns: AsyncProject instance
Raises: NotFoundError if project not found
Example:
project = await nexus.get_project("my-project")
create_project(project_name: str, metadata: Optional[dict] = None)
async def create_project(
project_name: str,
metadata: Optional[dict] = None,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncProject
Create a new project.
Parameters:
project_name(str): Name of the projectmetadata(Optionaldict): Project metadata
Returns: AsyncProject instance
Example:
project = await nexus.create_project("my-project", metadata={"key": "value"})
get_or_create_project(project_name: str, metadata: Optional[dict] = None) (convenience)
async def get_or_create_project(
project_name: str,
metadata: Optional[dict] = None,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncProject
Convenience method: get an existing project by name, or create it if it does not exist.
Parameters:
project_name(str): Name of the projectmetadata(Optionaldict): Project metadata (used only when creating)
Returns: AsyncProject instance
Example:
project = await nexus.get_or_create_project("my-project")
iter_projects(prefetch_pages: int = 1, page_size: int = 50)
async def iter_projects(
prefetch_pages: int = 1,
page_size: int = 50,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncIterator[ProjectInfo]
Iterate over all projects with transparent pagination.
Parameters:
prefetch_pages(int): Number of pages to prefetch (default: 1, max: 5)page_size(int): Number of items per page (default: 50, max: 1000)
Yields: ProjectInfo dictionaries
Example:
from miura.logging import get_logger
logger = get_logger(__name__)
async for project in nexus.iter_projects(prefetch_pages=2, page_size=100):
logger.info(f"Project: {project.name}")
AsyncProject
Async interface to a Nexus project.
Methods
list_collections()
async def list_collections(
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> list[CollectionInfo]
List all collections in this project.
Returns: List of CollectionInfo dictionaries
get_collection(collection_name: str)
async def get_collection(
collection_name: str,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncCollection
Get a collection by name.
Raises: NotFoundError if collection not found
create_collection(collection_name: str, schema: dict, metadata: Optional[dict] = None)
async def create_collection(
collection_name: str,
schema: dict,
metadata: Optional[dict] = None,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncCollection
Create a new collection.
get_or_create_collection(collection_name: str, schema: dict, metadata: Optional[dict] = None) (convenience)
async def get_or_create_collection(
collection_name: str,
schema: dict,
metadata: Optional[dict] = None,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncCollection
Convenience method: get an existing collection by name, or create it if it does not exist. Schema and metadata are used only when creating.
Parameters:
collection_name(str): Name of the collectionschema(dict): Schema definition (used only when creating)metadata(Optionaldict): Collection metadata (used only when creating)
Returns: AsyncCollection instance
Example:
collection = await project.get_or_create_collection(
collection_name="my-collection",
schema=schema,
)
iter_collections(prefetch_pages: int = 1, page_size: int = 50)
async def iter_collections(
prefetch_pages: int = 1,
page_size: int = 50,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncIterator[CollectionInfo]
Iterate over all collections with transparent pagination.
AsyncCollection
Async interface to a Nexus collection.
Methods
list_items(path: str = "/", include_dirs: bool = True, status_filter: Optional[str] = None, sort_by: Optional[str] = None, sort_order: str = "asc")
async def list_items(
path: str = "/",
include_dirs: bool = True,
status_filter: Optional[str] = None,
sort_by: Optional[str] = None,
sort_order: str = "asc",
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> list[AsyncCollectionItem]
List all items at the given path (single level). Loads all items into memory.
Single-level only (non-recursive): Returns only direct children at path; does not recurse into subdirectories. For recursive tree traversal use walk_items().
Returns: List of AsyncCollectionItem objects. For large directories, use iter_items() instead to stream with pagination.
Parameters:
path(str): Path within the collection (default: "/")include_dirs(bool): Include directories in results (default: True)status_filter(Optionalstr): Filter by status (e.g., "SUCCEEDED", "FAILED")sort_by(Optionalstr): Field to sort bysort_order(str): "asc" or "desc" (default: "asc")
Example:
from miura.logging import get_logger
logger = get_logger(__name__)
items = await collection.list_items(path="/data")
for item in items:
logger.info(f"{item.item_uri}: {item.file_size} bytes")
iter_items(path: str = "/", include_dirs: bool = True, status_filter: Optional[str] = None, sort_by: Optional[str] = None, sort_order: str = "asc", prefetch_pages: int = 1, page_size: int = 50)
async def iter_items(
path: str = "/",
include_dirs: bool = True,
status_filter: Optional[str] = None,
sort_by: Optional[str] = None,
sort_order: str = "asc",
prefetch_pages: int = 1,
page_size: int = 50,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncIterator[AsyncCollectionItem]
Iterate over items at the path with transparent pagination.
Single-level only (non-recursive): Yields only direct children at path; does not recurse into subdirectories (like pathlib.Path.iterdir()). For recursive tree traversal use walk_items().
Parameters:
prefetch_pages(int): Pages to prefetch (default: 1, max: 5)page_size(int): Items per page (default: 50, max: 1000)
Yields: AsyncCollectionItem objects (with .item_uri, .file_size, .is_folder, etc.)
Example:
from miura.logging import get_logger
logger = get_logger(__name__)
async for item in collection.iter_items(path="/data", sort_by="item_uri"):
logger.info(f"Item: {item.item_uri} ({item.file_size} bytes)")
walk_items(path: str = "/", max_depth: Optional[int] = None, topdown: bool = True, include_dirs: bool = True, ...)
async def walk_items(
path: str = "/",
max_depth: Optional[int] = None,
topdown: bool = True,
include_dirs: bool = True,
status_filter: Optional[str] = None,
sort_by: Optional[str] = None,
sort_order: str = "asc",
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> AsyncIterator[tuple[AsyncCollectionItem, int]]
Recursively walk the collection tree, yielding (item, depth) for each file and folder. Like os.walk(). For single-level iteration use iter_items().
Parameters:
path(str): Starting path (default: "/")max_depth(Optionalint): Maximum depth (None = unlimited)topdown(bool): If True, yield directory before its contents; if False, after (bottom-up)include_dirs(bool): Include directories (default: True)
Yields: Tuples (AsyncCollectionItem, depth) where depth is 0 at root, 1 one level down, etc.
Example:
async for item, depth in collection.walk_items(path="/"):
logger.info(f"{' ' * depth}{item.item_uri}")
upload(datasource, mode: UploadMode = UploadMode.APPEND)
from miura.sdk import UploadMode
async def upload(
datasource: LocalDataSource,
mode: UploadMode = UploadMode.APPEND,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> UploadResult
Upload Modes:
UploadMode.APPEND(default): Adds new items without deleting existing ones (safe default)UploadMode.REPLACE: Deletes all existing items before uploading (destructive)
Upload data from a datasource to this collection.
Note: Uses internal defaults for data organization.
get_item(item_path: str)
async def get_item(
item_path: str,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> Optional[AsyncCollectionItem]
Get a specific file or folder from this collection by its path.
Parameters:
item_path(str): Path to the item within the collection (e.g.,/data/file.txtor/models/)
Returns: AsyncCollectionItem if found, None if not found
Note: Folders should have a trailing slash (e.g., /models/), files should not.
Example:
# Get a folder
folder = await collection.get_item("/models/")
if folder:
logger.info(f"Found folder: {folder.name}")
download(path: str, local_path: str, confirm: bool = True)
async def download(
path: str,
local_path: str,
confirm: bool = True,
retry_policy: Optional[RetryPolicy] = None,
timeouts: Optional[Timeouts] = None,
hooks: Optional[ApiHooks] = None,
) -> DownloadResult
Download files from this collection to local storage.
Note: Uses internal defaults for data organization.
Sync API (miura)
The sync API is a compatibility layer that wraps the async API. Use it for synchronous code or Jupyter notebooks.
from miura import Nexus, Project, Collection
Note: Imports work without side effects (no network calls on import).
Jupyter Note: The sync API works in Jupyter notebooks using a background event loop.
Nexus
Synchronous wrapper around AsyncNexus.
with Nexus( config: OptionalClientConfig = None, api_key: Optionalstr = None, verify_ssl: bool = True, default_timeout: Optionalfloat = 30.0, max_workers: Optionalint = None, connection_limit: Optionalint = None, connection_limit_per_host: Optionalint = None, ) as nexus: # Use nexus here
Note: If config is not provided, configuration is loaded from environment (e.g. ClientConfig.from_env()). api_key and other args override env values.
Methods
All methods mirror the async API but are synchronous:
list_projects() -> list[Project]get_project(project_name: str) -> Projectcreate_project(project_name: str) -> Projectget_or_create_project(project_name: str, metadata=None) -> Project— (convenience) get existing or createdelete_project(project_name: str) -> None- Delete a project by name and all associated resourcesiter_projects(prefetch_pages=1, page_size=50) -> Iterator[ProjectInfo]get(path: str) -> Project | Collection | CollectionItem- Path-based navigationdownload(path: str, local_path: str, confirm=True) -> dictclose() -> None
get(path: str)
Get a project, collection, or collection item using hierarchical path-based navigation.
Parameters:
path(str): Hierarchical path:"project"→ returnsProject"project/collection"→ returnsCollection"project/collection/item/path"→ returnsCollectionItem
Returns: Project, Collection, or CollectionItem depending on path depth
Raises: ValueError if path is empty or resource not found
Example:
Get item (file)
item = nexus.get("my-project/my-collection/data/file.txt") result = item.download("./downloads/", confirm=False)
Delete a project by name
nexus.delete_project("my-project")
Note: This operation cannot be undone. All associated collections and data will be deleted.
Note: Cascade deletion is always enabled. When you delete a project, all associated collections, data, and resources are automatically deleted.
Project
Synchronous wrapper around AsyncProject.
Methods
list_collections() -> list[dict]get_collection(collection_name: str) -> Collectioncreate_collection(name: str, schema: dict, metadata=None) -> Collectionget_or_create_collection(name: str, schema: dict, metadata=None) -> Collection— (convenience) get existing or createiter_collections(prefetch_pages=1, page_size=50) -> Iterator[dict]update_project(metadata=None) -> Project(NotImplementedError)delete_project() -> None- Delete this project and all associated resourceshelp() -> None- Display help information
delete_project()
Delete this project and all associated resources (collections, data, etc.).
Returns: None
Raises:
NotFoundError: If project is not foundNetworkError: If network request failsPermissionError: If authentication fails
Example:
Collection
Synchronous wrapper around AsyncCollection.
Methods
list_items(path="/", include_dirs=True, status_filter=None, sort_by=None, sort_order="asc") -> list[CollectionItem]— single-level onlyiter_items(path="/", sort_by=None, sort_order="asc", prefetch_pages=1, page_size=50) -> Iterator[CollectionItem]— single-level onlywalk_items(path="/", max_depth=None, topdown=True, include_dirs=True) -> Iterator[tuple[CollectionItem, int]]— recursive tree walkget_item(item_path: str) -> Optional[CollectionItem]- Get specific item by pathupload(datasource, mode=UploadMode.APPEND) -> dictdownload(path: str, local_path: str, confirm=True) -> dicthelp() -> None- Display help information
get_item(item_path: str)
Get a specific file or folder from this collection by its path.
Parameters:
item_path(str): Path to the item within the collection (e.g.,/data/file.txtor/models/)
Returns: CollectionItem if found, None if not found
Note: Folders should have a trailing slash (e.g., /models/), files should not.
Example:
Get a folder
folder = collection.get_item("/models/") if folder: result = folder.download("./downloads/models/", confirm=False)
update_collection(metadata=None) -> Collection(NotImplementedError)delete_collection() -> None- Delete this collection and all associated lakehouse datahelp() -> None- Display help information
Exception Hierarchy
All exceptions derive from NexusError and include error codes, reasons, and details.
NexusError
Base exception for all Miura API errors.
Attributes:
code(str): Error code constant fromErrorCodeclassreason(str): Human-readable error message with remediation hintdetails(dict): Optional dictionary with additional context
Example:
from miura.logging import get_logger
logger = get_logger(__name__)
try:
project = await nexus.get_project("nonexistent")
except NexusError as e:
logger.error(f"Error code: {e.code}")
logger.error(f"Reason: {e.reason}")
logger.error(f"Details: {e.details}")
Specific Exceptions
NotFoundError
Raised when a requested resource is not found.
Error Code: ErrorCode.NOT_FOUND
Example:
from miura.sdk import NotFoundError
from miura.logging import get_logger
logger = get_logger(__name__)
try:
project = await nexus.get_project("nonexistent")
except NotFoundError as e:
logger.error(f"Project not found: {e.reason}")
ValidationError
Raised when input validation fails.
Error Code: ErrorCode.VALIDATION_ERROR
PermissionError
Raised when authentication or authorization fails.
Error Code: ErrorCode.PERMISSION_DENIED
NetworkError
Raised when network requests fail (retryable).
Error Code: ErrorCode.NETWORK_ERROR
TimeoutError
Raised when operations timeout.
Error Code: ErrorCode.TIMEOUT
CancelledError
Raised when operations are cancelled.
Error Code: ErrorCode.CANCELLED
ConflictError
Raised when resource conflicts occur (e.g., 409).
Error Code: ErrorCode.CONFLICT
Error Codes
All error codes are defined in the ErrorCode class:
| Constant | Value | Description |
|---|---|---|
ErrorCode.NOT_FOUND | "not_found" | Resource not found |
ErrorCode.VALIDATION_ERROR | "validation_error" | Input validation failed |
ErrorCode.PERMISSION_DENIED | "permission_denied" | Authentication/authorization failed |
ErrorCode.NETWORK_ERROR | "network_error" | Network request failed (retryable) |
ErrorCode.TIMEOUT | "timeout" | Operation timed out |
ErrorCode.CANCELLED | "cancelled" | Operation was cancelled |
ErrorCode.CONFLICT | "conflict" | Resource conflict (e.g., 409) |
Usage:
from miura.sdk import ErrorCode, NotFoundError
from miura.logging import get_logger
logger = get_logger(__name__)
if e.code == ErrorCode.NOT_FOUND:
logger.error("Resource not found")
HTTP→SDK Error Mapping
HTTP status codes are automatically mapped to SDK exceptions:
| HTTP Status | Exception Class | Error Code | Retryable |
|---|---|---|---|
| 400 | ValidationError | VALIDATION_ERROR | No |
| 401 | PermissionError | PERMISSION_DENIED | No |
| 403 | PermissionError | PERMISSION_DENIED | No |
| 404 | NotFoundError | NOT_FOUND | No |
| 409 | ConflictError | CONFLICT | No |
| 429 | NetworkError | NETWORK_ERROR | Yes |
| 500 | NetworkError | NETWORK_ERROR | Yes |
| 502 | NetworkError | NETWORK_ERROR | Yes |
| 503 | NetworkError | NETWORK_ERROR | Yes |
| 504 | TimeoutError | TIMEOUT | Yes |
Example:
Retry & Timeout Configuration
Retry Policy
Configure retry behavior with RetryPolicy:
from miura.sdk import RetryPolicy
retry_policy = RetryPolicy(
max_retries=3,
initial_backoff=1.0,
max_backoff=60.0,
jitter=True,
)
async with AsyncNexus(retry_policy=retry_policy) as nexus: projects = await nexus.list_projects()
Attributes:
max_retries(int): Maximum retry attempts (default: 3)initial_backoff(float): Initial backoff in seconds (default: 1.0)max_backoff(float): Maximum backoff in seconds (default: 60.0)jitter(bool): Apply jitter to backoff (default: True)
Per-Call Override:
Timeouts
Configure timeout values with Timeouts:
from miura.sdk import Timeouts
timeouts = Timeouts(
connect=10.0, # Connection timeout
read=30.0, # Read timeout
total=60.0, # Total timeout
)
async with AsyncNexus(timeouts=timeouts) as nexus:
projects = await nexus.list_projects()
Note: Timeout integration with HTTPClient is not yet complete (TODO).
Per-Call Override:
Override timeouts for specific call
projects = await nexus.list_projects( timeouts=Timeouts(total=120.0) )
Precedence
Policy precedence: per-call > client > transport
- Per-call overrides (highest priority)
- Client-level configuration
- Transport defaults (lowest priority)
Logging Hooks
Configure callback hooks for observability:
from miura.sdk import ApiHooks
from miura.logging import get_logger
logger = get_logger(__name__)
def on_request(metadata: dict) -> None:
logger.debug(f"Request: {metadata['method']}")
def on_response(metadata: dict) -> None:
logger.debug(f"Response: {metadata['method']} - {metadata['status']}")
def on_retry(metadata: dict) -> None: logger.debug(f"Retry: {metadata'method'} (attempt {metadata'attempt'})")
hooks = ApiHooks(
on_request=on_request,
on_response=on_response,
on_retry=on_retry,
)
async with AsyncNexus(hooks=hooks) as nexus: projects = await nexus.list_projects()
Per-Call Override:
Override hooks for specific call
projects = await nexus.list_projects( hooks=ApiHooks(on_request=custom_on_request) )
Note: Sync API doesn't yet support hooks in constructor (TODO).
See Hooks Documentation for detailed information.
Iterator Backpressure Knobs
Iterators support backpressure controls to manage memory usage:
Example
from miura.logging import get_logger
logger = get_logger(__name__)
Prefetch 2 pages, 100 items per page
async for project in nexus.iter_projects(prefetch_pages=2, page_size=100): logger.info(f"Project: {project.name} ({project.uuid})")
Deterministic Pagination
For deterministic pagination, provide sort_by parameter:
from miura.logging import get_logger
logger = get_logger(__name__)
Deterministic ordering
async for item in collection.iter_items(path="/", sort_by="name", sort_order="asc"): logger.info(f"Item: {item.item_uri} ({item.file_size} bytes)")
Note: Without sort_by, results may be reordered if backend changes.
Path Navigation
The get() method supports only 2 levels:
nexus.get("project")→ ReturnsProjectnexus.get("project/collection")→ ReturnsCollectionnexus.get("project/collection/extra")→ RaisesValueError
help() Method
Display help information about available methods:
nexus.help()
project.help()
collection.help()
Complete Examples
Async API Example
import asyncio
from miura.api import AsyncNexus
from miura.sdk import NotFoundError
from miura.logging import get_logger
logger = get_logger(__name__)
async def main():
async with AsyncNexus() as nexus:
# List projects
projects = await nexus.list_projects()
logger.info(f"Found {len(projects)} project(s)")
# Get or create project
try:
project = await nexus.get_project("my-project")
logger.info(f"Retrieved project: {project.name}")
except NotFoundError:
project = await nexus.create_project("my-project")
logger.info(f"Created project: {project.name}")
# Get collection
collection = await project.get_collection("my-collection")
logger.info(f"Retrieved collection: {collection.name}")
# List items
items = await collection.list_items(path="/")
logger.info(f"Found {len(items.get('items', []))} item(s)")
# Iterate items
async for item in collection.iter_items(path="/"):
logger.info(f"Item: {item.item_uri} ({item.file_size} bytes)")
asyncio.run(main())
Sync API Example
from miura import Nexus
from miura.sdk import NotFoundError
from miura.logging import get_logger
logger = get_logger(__name__)
with Nexus() as nexus:
# List projects
projects = nexus.list_projects()
logger.info(f"Found {len(projects)} project(s)")
# Get or create project
try:
project = nexus.get_project("my-project")
logger.info(f"Retrieved project: {project.name}")
except NotFoundError:
project = nexus.create_project("my-project")
logger.info(f"Created project: {project.name}")
# Get collection
collection = project.get_collection("my-collection")
logger.info(f"Retrieved collection: {collection.name}")
# List items
items = collection.list_items(path="/")
logger.info(f"Found {len(items.get('items', []))} item(s)")
# Iterate items
for item in collection.iter_items(path="/"):
logger.info(f"Item: {item.item_uri} ({item.file_size} bytes)")
Type Definitions (miura.types)
Type definitions are available from multiple locations for convenience. The canonical location is miura.types, which is independent of whether you're using the sync or async API.
Import
Types can be imported from any of these locations:
Shared types (datasources, exceptions, policies, schema, types) from miura.sdk
from miura.sdk import CollectionItemInfo, ProjectInfo
Recommendation: Use miura.sdk for type hints and shared utilities; it is independent of API style (async vs sync) and keeps your imports consistent.
Available Types
ProjectInfo: Information about a Nexus projectCollectionInfo: Information about a Nexus collectionCollectionItemInfo: Information about a collection item (file or folder)AsyncCollectionItem/CollectionItem: Item with collection context (wrapsCollectionItemInfo); provides.download(),.item_uri,.file_size, etc.list_items()returns a list ofAsyncCollectionItem(async) orCollectionItem(sync); useiter_items()for paginated streaming.PaginationInfo: Pagination information for paginated responsesUploadResult: Result of an upload operationDownloadResult: Result of a download operation
Example:
from miura.sdk import CollectionItemInfo, ProjectInfo
from miura.api import AsyncNexus
async def process_items(nexus: AsyncNexus, project: ProjectInfo):
collection = await project.get_collection("my-collection")
items = await collection.list_items()
for item in items:
# item is an AsyncCollectionItem
print(f"Item: {item.name} ({item.file_size} bytes)")
Additional Resources
- Nexus Quick Start - Progressive getting started guide
- Hooks Documentation - Detailed hooks documentation