API Reference
Nexus API Reference
This document provides comprehensive documentation for the high-level Nexus API classes and methods. The API is designed to provide a synchronous, user-friendly interface to the Nexus platform while hiding the complexity of async operations.
Nexus
The main client class that provides a synchronous facade over the async NexusClient.
Class Overview
class Nexus:
"""
Synchronous facade over the async NexusClient for easy integration.
This class provides a user-friendly synchronous interface to the Nexus platform,
hiding the complexity of async operations while maintaining full functionality.
It automatically handles environment configuration, logging, and resource cleanup.
"""
Constructor
def __init__(
self,
env_path: Optional[Path] = None,
logger_name: str = "nexus-sync",
logger_mode: str = "quiet",
verify_ssl: bool = False,
default_timeout: Optional[float] = 30.0,
):
Parameters:
env_path(OptionalPath): Path to environment configuration file. If None, uses hierarchical loading.logger_name(str): Name for the logger instance (default: "nexus-sync")logger_mode(str): Logging mode - "quiet", "normal", or "verbose" (default: "quiet")verify_ssl(bool): Whether to verify SSL certificates (default: False)default_timeout(float): Default timeout for operations in seconds (default: 30.0)
Environment Loading: The client automatically loads environment variables using AppConfig's hierarchical loading system:
.env.localin current directory.env.localin package rootnexus_example.envin examples directory- OS environment variables
Methods
list_projects()
def list_projects(self) -> List[Dict[str, Any]]
List all available projects for the authenticated user.
Returns:
- List of project dictionaries containing project information including name, UUID, owner, creation date, and metadata.
Raises:
Exception: If the API request fails or authentication errors occur
get_project(project_name: str)
def get_project(self, project_name: str) -> Project
Get a project by name, returning a Project wrapper object.
Parameters:
project_name(str): Name of the project to retrieve
Returns:
Projectwrapper object for the specified project
Raises:
ValueError: If the project with the given name is not foundException: If the API request fails or authentication errors occur
create_project(project_name: str)
def create_project(self, project_name: str) -> Project
Create a new project or return existing one, returning a Project wrapper object.
Parameters:
project_name(str): Name of the project to create or retrieve
Returns:
Projectwrapper object for the created or existing project
Raises:
Exception: If the API request fails or authentication errors occur
Note: If a project with the same name already exists, the existing project will be returned instead of creating a new one.
get(path: str)
def get(self, path: str) -> Union[Project, Collection, Group]
Get a project, collection, or group using a hierarchical path.
Parameters:
path(str): Hierarchical path in format "project/collection/group"- "project" → returns Project object
- "project/collection" → returns Collection object
- "project/collection/group" → returns Group object
Returns:
- Project, Collection, or Group wrapper object depending on path depth
Examples:
nexus.get("my-project") # -> Project
nexus.get("my-project/my-collection") # -> Collection
nexus.get("my-project/my-collection/raw") # -> Group
download(path: str, local_path: str, confirm: bool = True)
def download(self, path: str, local_path: str, confirm: bool = True) -> Dict[str, Any]
Download files from a hierarchical path to local storage.
Parameters:
path(str): Hierarchical path in format "project/collection/group/path/to/files"- "project/collection" → downloads entire collection
- "project/collection/group" → downloads entire group
- "project/collection/group/folder1/folder2" → downloads specific folder
local_path(str): Local directory to download files to. Will be created if it doesn't exist.confirm(bool): Whether to ask for confirmation before downloading (default: True)
Returns: Dictionary containing download statistics:
status: "completed", "cancelled", or "error"files_downloaded: Number of files successfully downloadedtotal_size: Total size of downloaded data in byteserrors: List of any errors encountered
Raises:
ValueError: If the path format is invalid or path doesn't existException: If download fails or authentication errors occur
Examples:
nexus.download("my-project/my-collection", "./downloads/")
nexus.download("my-project/my-collection/raw", "./downloads/")
nexus.download("my-project/my-collection/raw/data", "./downloads/")
close()
def close(self) -> None
Close the client and clean up resources.
This method properly shuts down the Nexus client, including:
- Closing the underlying async client
- Shutting down the async runner
- Cleaning up any remaining resources
Note: This method should be called when the client is no longer needed to ensure proper cleanup of resources. The client can be used again after calling close() by creating a new instance.
Project
User-friendly wrapper for a Nexus project.
Class Overview
class Project:
"""
User-friendly wrapper for a Nexus project.
This class provides a synchronous interface to project operations,
including listing collections, creating new collections, and displaying
project information. It acts as a facade over the async NexusClient
to provide a simpler, synchronous API.
"""
Attributes
uuid: Unique identifier for the projectname: Human-readable name of the projectowner_id: ID of the project ownercreated_at: Project creation timestampupdated_at: Last modification timestampmetadata: Dictionary containing project metadata
Methods
list_collections()
def list_collections(self) -> List[Dict[str, Any]]
List all collections in this project.
Returns:
- List of collection dictionaries containing collection information including name, UUID, creation date, and metadata.
Raises:
Exception: If the API request fails or authentication errors occur
get_collection(collection_name: str)
def get_collection(self, collection_name: str) -> Collection
Get a collection by name, returning a Collection wrapper object.
Parameters:
collection_name(str): Name of the collection to retrieve
Returns:
Collectionwrapper object for the specified collection
Raises:
ValueError: If the collection with the given name is not foundException: If the API request fails or authentication errors occur
create_collection(name: str, schema: Dict[str, Any], metadata: Dict[str, Any] = None)
def create_collection(
self,
name: str,
schema: Dict[str, Any],
metadata: Dict[str, Any] = None
) -> Collection
Create a new collection in this project.
Parameters:
name(str): Name of the collectionschema(Dictstr, Any): Schema definition for the collectionmetadata(Dictstr, Any, optional): Optional metadata for the collection
Returns:
Collectionwrapper object for the created or existing collection
print_info()
def print_info(self) -> None
Print comprehensive information about the project and its collections.
Collection
User-friendly wrapper for a Nexus collection.
Class Overview
class Collection:
"""
User-friendly wrapper for a Nexus collection.
This class provides a synchronous interface to collection operations,
including listing groups, uploading data, and displaying collection
information. It acts as a facade over the async NexusClient to provide
a simpler, synchronous API.
"""
Attributes
uuid: Unique identifier for the collectionname: Human-readable name of the collectioncreated_at: Collection creation timestampupdated_at: Last modification timestampmetadata: Dictionary containing collection metadataproject_uuid: UUID of the parent projectproject_name: Name of the parent project
Methods
get_groups()
def get_groups(self) -> Dict[str, Any]
Get all groups for this collection.
Returns:
- Dictionary containing group information with 'groups' key containing a list of group names available in this collection.
Raises:
Exception: If the API request fails or authentication errors occur
get_group(group_name: str)
def get_group(self, group_name: str) -> Group
Get a group for this collection, returning a Group wrapper object.
Parameters:
group_name(str): Name of the group to retrieve
Returns:
Groupwrapper object for the specified group
Raises:
Exception: If the API request fails or authentication errors occur
get_items(group_name: str, path: str = "/", page: int = 1, limit: Optional[int] = None)
def get_items(
self,
group_name: str,
path: str = "/",
page: int = 1,
limit: Optional[int] = None
) -> Dict[str, Any]
Get items in this collection for a specific group.
Parameters:
group_name(str): Name of the group to querypath(str): Path within the group to browse (default: "/" for root)page(int): Page number for pagination (default: 1)limit(int, optional): Maximum number of items per page (default: None for all)
Returns:
- Dictionary containing items data with 'data' and 'pagination' keys
Raises:
Exception: If the API request fails or authentication errors occur
upload(datasource, group_name: str = "raw", create_new_version: bool = True)
def upload(
self,
datasource: LocalDataSource,
group_name: str = "raw",
create_new_version: bool = True
) -> Dict[str, Any]
Upload data from a datasource to this collection.
Parameters:
datasource(LocalDataSource): LocalDataSource object containing the data to uploadgroup_name(str): Name of the group to upload to (default: "raw")create_new_version(bool): Whether to create a new version (default: True)
Returns:
- Upload result dictionary with statistics
print_info()
def print_info(self) -> None
Print comprehensive information about the collection and its groups.
Group
User-friendly wrapper for a Nexus group.
Class Overview
class Group:
"""
User-friendly wrapper for a Nexus group.
This class provides a synchronous interface to group operations,
including browsing items, getting download URLs, and displaying
group information. It acts as a facade over the async NexusClient
to provide a simpler, synchronous API.
"""
Attributes
group_name: Name of the groupcollection_uuid: UUID of the parent collectioncollection_name: Name of the parent collectionproject_uuid: UUID of the parent projectproject_name: Name of the parent project
Methods
get_items(path: str = "/", page: int = 1, limit: Optional[int] = None)
def get_items(
self,
path: str = "/",
page: int = 1,
limit: Optional[int] = None
) -> Dict[str, Any]
Get items in this group at the specified path.
Parameters:
path(str): Path within the group to browse (default: "/" for root)page(int): Page number for pagination (default: 1)limit(int, optional): Maximum number of items per page (default: None for all)
Returns:
- Dictionary containing items data with 'data' and 'pagination' keys
Raises:
Exception: If the API request fails or authentication errors occur
get_download_urls(object_keys: List[str], page: int = 1, limit: Optional[int] = None)
def get_download_urls(
self,
object_keys: List[str],
page: int = 1,
limit: Optional[int] = None
) -> Dict[str, Any]
Get download URLs for specific items in this group.
Parameters:
object_keys(Liststr): List of object keys to get download URLs forpage(int): Page number for pagination (default: 1)limit(int, optional): Maximum number of items per page (default: None for all)
Returns:
- Dictionary containing download URL information
Raises:
Exception: If the API request fails or authentication errors occur
browse_tree(path: str = "/", max_depth: int = 3)
def browse_tree(self, path: str = "/", max_depth: int = 3) -> None
Browse the tree structure of items in this group.
Parameters:
path(str): Path within the group to browse (default: "/" for root)max_depth(int): Maximum depth to explore (default: 3)
print_info()
def print_info(self) -> None
Print comprehensive information about the group and its items.
LocalDataSource
Represents a local data source for collection uploads.
Class Overview
class LocalDataSource:
"""
Represents a local data source for collection uploads.
This class provides a simple wrapper around a local folder path
that can be used as a data source for collection operations. It
validates the path, provides file information, and supports
various file operations for upload preparation.
"""
Attributes
path: Resolved Path object pointing to the data source directoryexists: Boolean property indicating if the path existsis_directory: Boolean property indicating if the path is a directory
Constructor
def __init__(self, path: str):
Parameters:
path(str): Path to the local folder containing data files
Raises:
ValueError: If the path doesn't exist or is not a directory
Methods
get_file_count()
def get_file_count(self) -> int
Get the total number of files in the data source.
Returns:
- Total count of files in the data source directory and subdirectories. Returns 0 if the directory cannot be accessed.
get_file_list(pattern: str = "*")
def get_file_list(self, pattern: str = "*") -> List[str]
Get a list of files in the data source.
Parameters:
pattern(str): Glob pattern to match files (default: "*" for all files)
Returns:
- List of file paths relative to the data source root. Returns empty list if the directory cannot be accessed.
get_size_bytes()
def get_size_bytes(self) -> int
Get the total size of all files in the data source in bytes.
Returns:
- Total size of all files in the data source directory and subdirectories. Returns 0 if the directory cannot be accessed.
print_info()
def print_info(self) -> None
Print information about the data source.
Helper Functions
configure_env()
def configure_env(
env_vars: Optional[Dict[str, Any]] = None,
config_path: Optional[Union[str, Path]] = None,
update_existing: bool = True,
validate: bool = True,
) -> Path:
Configure Nexus environment variables with automatic AppConfig field extraction.
Parameters:
env_vars(OptionalDictstr, Any): Dictionary of environment variables to set/updateconfig_path(OptionalUnionstr, Path): Path to save the config file. If None, uses package root/.env.localupdate_existing(bool): If True, preserves existing values in the file and only updates specified variablesvalidate(bool): If True, validates AppConfig fields against the schema
Returns:
- Path to the created/updated config file
Raises:
ValueError: If validation fails, invalid parameters provided, or conflicting variable names are detectedFileNotFoundError: If config_path directory doesn't exist and cannot be createdPermissionError: If the config file cannot be written due to permissions
Examples:
# Basic configuration
config_path = miura.configure_env({
'KC_BASE_URL': 'https://prod.auth.miurasimulation.com',
'nx_max_workers': 8
})
# Update existing configuration
miura.configure_env({
'nx_max_workers': 16
}, update_existing=True)
# Custom configuration file
miura.configure_env(
env_vars={'KC_BASE_URL': 'https://test.auth.example.com'},
config_path='custom.env'
)
Usage Examples
Basic Usage
from miura.api.nexus import Nexus
from miura.api.datasources import LocalDataSource
# Initialize the client
nexus = Nexus()
# List projects
projects = nexus.list_projects()
print(f"Found {len(projects)} projects")
# Get a specific project
project = nexus.get_project("my-project")
# List collections in the project
collections = project.list_collections()
print(f"Project has {len(collections)} collections")
# Get a specific collection
collection = project.get_collection("my-collection")
# List groups in the collection
groups = collection.get_groups()
print(f"Collection has groups: {groups['groups']}")
# Get a specific group
group = collection.get_group("raw")
# Browse items in the group
items = group.get_items()
print(f"Group has {len(items['data']['items'])} items")
# Close the client
nexus.close()
Upload Data
from miura.api.nexus import Nexus
from miura.api.datasources import LocalDataSource
# Initialize the client
nexus = Nexus()
# Create a data source
datasource = LocalDataSource("/path/to/local/data")
# Get or create a project
project = nexus.create_project("my-project")
# Create a collection
collection = project.create_collection(
name="my-collection",
schema={"type": "object", "properties": {"file_type": {"type": "string"}}},
metadata={"description": "My data collection"}
)
# Upload data to the collection
result = collection.upload(datasource, group_name="raw")
print(f"Upload completed: {result['files_uploaded']} files uploaded")
# Close the client
nexus.close()
Download Data
from miura.api.nexus import Nexus
# Initialize the client
nexus = Nexus()
# Download entire collection
result = nexus.download("my-project/my-collection", "./downloads/")
print(f"Download completed: {result['files_downloaded']} files downloaded")
# Download specific group
result = nexus.download("my-project/my-collection/raw", "./downloads/")
print(f"Download completed: {result['files_downloaded']} files downloaded")
# Download specific subdirectory
result = nexus.download("my-project/my-collection/raw/data", "./downloads/")
print(f"Download completed: {result['files_downloaded']} files downloaded")
# Close the client
nexus.close()
Hierarchical Navigation
from miura.api.nexus import Nexus
# Initialize the client
nexus = Nexus()
# Navigate using hierarchical paths
project = nexus.get("my-project") # Returns Project object
collection = nexus.get("my-project/my-collection") # Returns Collection object
group = nexus.get("my-project/my-collection/raw") # Returns Group object
# All objects support the same interface
project.print_info()
collection.print_info()
group.print_info()
# Close the client
nexus.close()
Error Handling
The API provides comprehensive error handling with meaningful error messages:
- Authentication Errors: Check your credentials and API key configuration
- Network Errors: Verify your internet connection and API endpoint
- Validation Errors: Ensure your input parameters are correct
- Permission Errors: Verify you have access to the requested resources
All methods that can fail will raise appropriate exceptions with descriptive error messages to help with debugging.
Best Practices
- Always close the client: Call
nexus.close()when done to ensure proper cleanup - Use context managers: Consider using try/finally blocks for proper resource management
- Handle exceptions: Wrap API calls in try/except blocks for robust error handling
- Check return values: Verify that operations completed successfully before proceeding
- Use hierarchical paths: Leverage the
get()method for clean navigation - Configure logging: Set appropriate logging levels for your use case
This documentation covers the high-level API. For more advanced usage and low-level operations, refer to the underlying NexusClient documentation.