API Reference

Complete documentation of all classes and methods

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.

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.

python
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

python
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 configuration
  • timeouts (OptionalTimeouts): Timeout configuration
  • hooks (OptionalApiHooks): Callback hooks for observability
  • verify_ssl (bool): Whether to verify SSL certificates (default: True)

Methods

python
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:

python
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)
python
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:

python
project = await nexus.get_project("my-project")
create_project(project_name: str, metadata: Optional[dict] = None)
python
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 project
  • metadata (Optionaldict): Project metadata

Returns: AsyncProject instance

Example:

python
project = await nexus.create_project("my-project", metadata={"key": "value"})
get_or_create_project(project_name: str, metadata: Optional[dict] = None) (convenience)
python
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 project
  • metadata (Optionaldict): Project metadata (used only when creating)

Returns: AsyncProject instance

Example:

python
project = await nexus.get_or_create_project("my-project")
iter_projects(prefetch_pages: int = 1, page_size: int = 50)
python
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:

python
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()
python
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)
python
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)
python
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)
python
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 collection
  • schema (dict): Schema definition (used only when creating)
  • metadata (Optionaldict): Collection metadata (used only when creating)

Returns: AsyncCollection instance

Example:

python
collection = await project.get_or_create_collection(
    collection_name="my-collection",
    schema=schema,
)
iter_collections(prefetch_pages: int = 1, page_size: int = 50)
python
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")
python
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 by
  • sort_order (str): "asc" or "desc" (default: "asc")

Example:

python
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)
python
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:

python
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, ...)
python
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:

python
async for item, depth in collection.walk_items(path="/"):
    logger.info(f"{'  ' * depth}{item.item_uri}")
upload(datasource, mode: UploadMode = UploadMode.APPEND)
python
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)
python
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.txt or /models/)

Returns: AsyncCollectionItem if found, None if not found

Note: Folders should have a trailing slash (e.g., /models/), files should not.

Example:

python
# 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)
python
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.

python
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) -> Project
  • create_project(project_name: str) -> Project
  • get_or_create_project(project_name: str, metadata=None) -> Project(convenience) get existing or create
  • delete_project(project_name: str) -> None - Delete a project by name and all associated resources
  • iter_projects(prefetch_pages=1, page_size=50) -> Iterator[ProjectInfo]
  • get(path: str) -> Project | Collection | CollectionItem - Path-based navigation
  • download(path: str, local_path: str, confirm=True) -> dict
  • close() -> None
get(path: str)

Get a project, collection, or collection item using hierarchical path-based navigation.

Parameters:

  • path (str): Hierarchical path:
    • "project" → returns Project
    • "project/collection" → returns Collection
    • "project/collection/item/path" → returns CollectionItem

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) -> Collection
  • create_collection(name: str, schema: dict, metadata=None) -> Collection
  • get_or_create_collection(name: str, schema: dict, metadata=None) -> Collection(convenience) get existing or create
  • iter_collections(prefetch_pages=1, page_size=50) -> Iterator[dict]
  • update_project(metadata=None) -> Project (NotImplementedError)
  • delete_project() -> None - Delete this project and all associated resources
  • help() -> None - Display help information
delete_project()

Delete this project and all associated resources (collections, data, etc.).

Returns: None

Raises:

  • NotFoundError: If project is not found
  • NetworkError: If network request fails
  • PermissionError: 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 only
  • iter_items(path="/", sort_by=None, sort_order="asc", prefetch_pages=1, page_size=50) -> Iterator[CollectionItem] — single-level only
  • walk_items(path="/", max_depth=None, topdown=True, include_dirs=True) -> Iterator[tuple[CollectionItem, int]] — recursive tree walk
  • get_item(item_path: str) -> Optional[CollectionItem] - Get specific item by path
  • upload(datasource, mode=UploadMode.APPEND) -> dict
  • download(path: str, local_path: str, confirm=True) -> dict
  • help() -> 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.txt or /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 data
  • help() -> 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 from ErrorCode class
  • reason (str): Human-readable error message with remediation hint
  • details (dict): Optional dictionary with additional context

Example:

python
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:

python
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:

ConstantValueDescription
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:

python
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 StatusException ClassError CodeRetryable
400ValidationErrorVALIDATION_ERRORNo
401PermissionErrorPERMISSION_DENIEDNo
403PermissionErrorPERMISSION_DENIEDNo
404NotFoundErrorNOT_FOUNDNo
409ConflictErrorCONFLICTNo
429NetworkErrorNETWORK_ERRORYes
500NetworkErrorNETWORK_ERRORYes
502NetworkErrorNETWORK_ERRORYes
503NetworkErrorNETWORK_ERRORYes
504TimeoutErrorTIMEOUTYes

Example:

Retry & Timeout Configuration

Retry Policy

Configure retry behavior with RetryPolicy:

python
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:

python
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

  1. Per-call overrides (highest priority)
  2. Client-level configuration
  3. Transport defaults (lowest priority)

Logging Hooks

Configure callback hooks for observability:

python
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'})")

python
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

python
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:

python
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") → Returns Project
  • nexus.get("project/collection") → Returns Collection
  • nexus.get("project/collection/extra") → Raises ValueError

help() Method

Display help information about available methods:

python
nexus.help()
project.help()
collection.help()

Complete Examples

Async API Example

python
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

python
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 project
  • CollectionInfo: Information about a Nexus collection
  • CollectionItemInfo: Information about a collection item (file or folder)
  • AsyncCollectionItem / CollectionItem: Item with collection context (wraps CollectionItemInfo); provides .download(), .item_uri, .file_size, etc.
  • list_items() returns a list of AsyncCollectionItem (async) or CollectionItem (sync); use iter_items() for paginated streaming.
  • PaginationInfo: Pagination information for paginated responses
  • UploadResult: Result of an upload operation
  • DownloadResult: Result of a download operation

Example:

python
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

© 2026