This document provides comprehensive documentation for the high-level Nexus API classes and methods.

Async is Default: The primary API is async-first (miura.aio). 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.aio): 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.aio)

The async API is the primary, canonical interface. Use it for all new code.

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

Methods

async def list_projects( retry_policy: OptionalRetryPolicy = None, timeouts: OptionalTimeouts = None, hooks: OptionalApiHooks = None, ) -> listProjectInfo

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 project
• metadata (Optionaldict): Project metadata

Returns: AsyncProject instance

Example:

project = await nexus.create_project("my-project", metadata={"key": "value"})

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.

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 = "/", page: int = 1, page_size: int = 50, sort_by: Optional[str] = None, sort_order: str = "asc")

async def list_items(
    path: str = "/", 
    page: int = 1, 
    page_size: int = 50,
    sort_by: Optional[str] = None,
    sort_order: str = "asc",
    retry_policy: Optional[RetryPolicy] = None,
    timeouts: Optional[Timeouts] = None,
    hooks: Optional[ApiHooks] = None,
) -> ItemsResponse

List items in this collection with pagination.

Returns: ItemsResponse containing a list of CollectionItem objects and pagination information.

Parameters:

• path (str): Path within the collection (default: "/")
• page (int): Page number (1-based, default: 1)
• page_size (int): Items per page (default: 50)
• sort_by (Optionalstr): Field to sort by
• sort_order (str): "asc" or "desc" (default: "asc")

Note: Uses internal defaults for data organization and filtering.

Example:

from miura.logging import get_logger

logger = get_logger(__name__)

response = await collection.list_items(path="/data", page=1, page_size=10)
for item in response["items"]:
    logger.info(f"{item.item_uri}: {item.file_size} bytes")

iter_items(path: str = "/", sort_by: Optional[str] = None, sort_order: str = "asc", prefetch_pages: int = 1, page_size: int = 50)

async def iter_items(
    path: str = "/", 
    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[CollectionItem]

Iterate over all items with transparent pagination.

Parameters:

• prefetch_pages (int): Pages to prefetch (default: 1, max: 5)
• page_size (int): Items per page (default: 50, max: 1000)

Yields: CollectionItem objects

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)")

upload(datasource, mode: UploadMode = UploadMode.APPEND)

from miura.api 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.

async def get_item( item_path: str, retry_policy: OptionalRetryPolicy = None, timeouts: OptionalTimeouts = None, hooks: OptionalApiHooks = None, ) -> OptionalBoundCollectionItem

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: BoundCollectionItem 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}")

async def download( path: str, local_path: str, confirm: bool = True, retry_policy: OptionalRetryPolicy = None, timeouts: OptionalTimeouts = None, hooks: OptionalApiHooks = 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( env_path: OptionalPath = None, logger_name: str = "nexus-sync", logger_mode: str = "quiet", verify_ssl: bool = False, default_timeout: Optionalfloat = 30.0, ) as nexus: # Use nexus here

Note: retry_policy, timeouts, and hooks parameters are not yet supported in the constructor (TODO).

Methods

All methods mirror the async API but are synchronous:

• list_projects() -> list[dict]
• get_project(project_name: str) -> Project
• create_project(project_name: str) -> Project
• delete_project(project_id: str) -> None - Delete a project by ID and all associated resources
• iter_projects(prefetch_pages=1, page_size=50) -> Iterator[dict]
• get(path: str) -> Project | Collection | BoundCollectionItem - 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 BoundCollectionItem

Returns: Project, Collection, or BoundCollectionItem 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 ID

nexus.delete_project("project-uuid-12345")

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
• 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="/", page=1, page_size=50, sort_by=None, sort_order="asc") -> dict
• iter_items(path="/", sort_by=None, sort_order="asc", prefetch_pages=1, page_size=50) -> Iterator[CollectionItem]
• get_item(item_path: str) -> Optional[BoundCollectionItem] - 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: BoundCollectionItem 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:

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.api.exceptions 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:

Usage:

from miura.api.exceptions 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:

Example:

Retry & Timeout Configuration

Retry Policy

Configure retry behavior with RetryPolicy:

from miura.api.policies 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.api.policies 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:

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

help() Method

Display help information about available methods:

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

Complete Examples

Async API Example

import asyncio
from miura.aio import AsyncNexus
from miura.api.exceptions 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.api.exceptions 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)")

Schema Generation (miura.api)

The schema generation module provides functionality to automatically generate collection schemas from existing filesystem structures.

Import

from miura.api import generate_schema_from_path, SchemaGenOptions

generate_schema_from_path()

def generate_schema_from_path(
    path: str | Path,
    options: Optional[SchemaGenOptions] = None,
) -> List[Dict[str, Any]]

Generate a collection schema from an existing filesystem path.

Parameters:

• path (str | Path): Filesystem path to scan (file or directory)
• options (OptionalSchemaGenOptions): Optional configuration for schema generation

Returns: List of root-level schema node dictionaries. This list can be passed directly to project.create_collection() as the schema parameter.

Example:

from miura.api import generate_schema_from_path, SchemaGenOptions
from miura import Nexus

Generate schema from filesystem

schema = generate_schema_from_path("data/my-simulation")

Use with collection creation

with Nexus() as nexus: project = nexus.create_project("my-project") collection = project.create_collection( name="my-collection", schema=schema )

Raises:

• FileNotFoundError: If the path doesn't exist
• PermissionError: If access to the path is denied

@dataclass class SchemaGenOptions: min_files_for_pattern: int = 3 default_required: bool = True schema_name: Optionalstr = None similarity_threshold: float = 0.5 confidence_threshold: float = 0.75 max_samples: int = 100 ascii_words: bool = True strict_dates: bool = False preserve_extension: bool = True forbid_slash: bool = True

Configuration options for schema generation.

Parameters:

• min_files_for_pattern (int): Minimum number of similar files needed to infer a pattern (default: 3)
• default_required (bool): If True, files appearing in all folders are marked as required (default: True)
• schema_name (Optionalstr): Name for the generated schema (default: auto-generated from path)
• similarity_threshold (float): Minimum similarity threshold for pattern detection (default: 0.5)
• confidence_threshold (float): Minimum confidence threshold for accepting patterns (default: 0.75)
• max_samples (int): Maximum number of files to sample per directory (default: 100)
• ascii_words (bool): Use ASCII-only word patterns (default: True)
• strict_dates (bool): Enable strict date validation (default: False)
• preserve_extension (bool): Preserve file extensions in patterns (default: True)
• forbid_slash (bool): Reject names containing slashes (default: True)

Example:

from miura.api import generate_schema_from_path, SchemaGenOptions

options = SchemaGenOptions(
    min_files_for_pattern=2,
    default_required=False,
    schema_name="my-custom-schema"
)
schema = generate_schema_from_path("data/my-simulation", options=options)

Note: The old import path from miura.nexus.collection import generate_schema_from_path is no longer available. Use from miura.api import generate_schema_from_path instead.

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:

Also available from API-specific modules

from miura.api.types import CollectionItem, ProjectInfo # Sync API from miura.aio.types import CollectionItem, ProjectInfo # Async API

Recommendation: Use miura.types for type hints as it's independent of API style and makes your code more flexible.

Available Types

• ProjectInfo: Information about a Nexus project
• CollectionInfo: Information about a Nexus collection
• CollectionItem: Information about a collection item (file or folder)
• BoundCollectionItem: CollectionItem with collection context for convenience methods
• ItemsResponse: Response containing items with pagination information
• PaginationInfo: Pagination information for paginated responses
• UploadResult: Result of an upload operation
• DownloadResult: Result of a download operation

Example:

from miura.types import CollectionItem, ProjectInfo
from miura.aio 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.get("items", []):
        # item is a CollectionItem
        print(f"Item: {item.name} ({item.file_size} bytes)")

Additional Resources

• Nexus Quick Start - Progressive getting started guide
• Hooks Documentation - Detailed hooks documentation