kumiho

Kumiho Python Client Library.

Kumiho is a graph-native creative and AI asset management system that tracks revisions, relationships, and lineage without uploading original files to the cloud. This SDK provides a Pythonic interface to the Kumiho gRPC backend.

Getting Started:

The simplest way to use Kumiho is with the top-level functions that use a default client configured from your environment:

import kumiho

# Authenticate and configure (run once per session)
kumiho.auto_configure_from_discovery() # This can be avoided if env var KUMIHO_AUTO_CONFIGURE=1 is set

# Create a project
project = kumiho.create_project("my-vfx-project", "VFX assets for commercial")

# Create spaces and items
space = project.create_space("characters")
item = space.create_item("hero", "model")

# Create revisions and artifacts
revision = item.create_revision()
artifact = revision.create_artifact("main", "/path/to/hero.fbx")

For more control, you can create a client manually:

import kumiho

client = kumiho.connect(
    endpoint="localhost:50051",
    token="your-auth-token"
)

with kumiho.use_client(client):
    projects = kumiho.get_projects()

Key Concepts:

• Project: Top-level container for all assets and spaces.

• Space: Hierarchical folder structure within a project.

• Item: A versioned asset (model, texture, workflow, etc.).

• Revision: A specific iteration of an item with artifacts.

• Artifact: A file reference (path/URI) within a revision.

• Edge: A relationship between revisions (dependencies, references).

• Kref: A URI-based unique identifier for any Kumiho object.

Authentication:

Kumiho uses Firebase authentication. Run the CLI to log in:

kumiho-auth login

This caches credentials in ~/.kumiho/. Then use auto_configure_from_discovery() to bootstrap the client.

Environment Variables:

• KUMIHO_AUTO_CONFIGURE: Set to “1” to auto-configure on import.

• KUMIHO_AUTH_TOKEN: Override the authentication token.

• KUMIHO_CONTROL_PLANE_URL: Override the control plane URL.

• KUMIHO_ENDPOINT: Override the gRPC endpoint.

Example

Complete workflow example:

import kumiho

# Configure client from cached credentials
kumiho.auto_configure_from_discovery()

# Get or create project
project = kumiho.get_project("my-project")
if not project:
    project = kumiho.create_project("my-project", "My VFX project")

# Navigate to asset space
assets = project.get_space("assets") or project.create_space("assets")

# Create a new model item
model = assets.create_item("character", "model")

# Create first revision
v1 = model.create_revision(metadata={"author": "artist1"})
v1.create_artifact("mesh", "/projects/char/v1/mesh.fbx")
v1.create_artifact("textures", "/projects/char/v1/textures.zip")
v1.tag("approved")
v1.tag("published") # published tag is reserved tag within Kumiho as revision with immutable semantics

# Query by kref
item = kumiho.get_item("kref://my-project/assets/character.model")
revision = kumiho.get_revision("kref://my-project/assets/character.model?r=1")

# Search across project
models = kumiho.item_search(
    context_filter="my-project",
    kind_filter="model"
)

Note

Kumiho follows a “BYO Storage” philosophy—files remain on your local or network storage. Kumiho only tracks paths, metadata, and relationships in its graph database.

See also

• Kumiho documentation: https://docs.kumiho.cloud

• GitHub: https://github.com/kumihoclouds/kumiho-python

kumiho.__version__

The current version of the kumiho package.

Type:

str

kumiho.LATEST_TAG

Standard tag name for the latest revision.

Type:

str

kumiho.PUBLISHED_TAG

Standard tag name for published revisions.

Type:

str

class kumiho.KumihoObject

Bases: object

Base class for all high-level Kumiho domain objects.

This abstract base class provides common functionality shared by all Kumiho objects, including access to the client for making API calls.

All domain objects (Project, Space, Item, Revision, Artifact, Edge) inherit from this class.

_client

The client instance for making API calls (internal).

Note

This is an internal base class. Users typically interact with concrete subclasses like Project or Version.

__init__(client)

Initialize the Kumiho object with a client reference.

Parameters:

client (_Client) – The client instance for making API calls.

Return type:

None

exception kumiho.KumihoError

Bases: Exception

Base exception class for all Kumiho errors.

All custom exceptions raised by the Kumiho SDK inherit from this class, making it easy to catch all Kumiho-related errors.

Example:

import kumiho

try:
    project = kumiho.get_project("nonexistent")
except kumiho.KumihoError as e:
    print(f"Kumiho error: {e}")

class kumiho.Project

Bases: KumihoObject

A Kumiho project—the top-level container for assets.

Projects are the root of the Kumiho hierarchy. Each project has its own namespace for spaces and items, and manages access control and settings independently.

Projects support both public and private access modes, allowing you to share assets publicly or restrict them to authenticated users.

project_id

The unique identifier for this project.

Type:

str

name

The URL-safe name of the project (e.g., “film-2024”).

Type:

str

description

Human-readable description of the project.

Type:

str

created_at

ISO timestamp when the project was created.

Type:

Optional[str]

updated_at

ISO timestamp of the last update.

Type:

Optional[str]

deprecated

Whether the project is deprecated (soft-deleted).

Type:

bool

allow_public

Whether anonymous read access is enabled.

Type:

bool

Example

Basic project operations:

import kumiho

# Get existing project
project = kumiho.get_project("my-project")

# Create spaces
assets = project.create_space("assets")
shots = project.create_space("shots")

# Navigate to nested spaces
char_space = project.get_space("assets/characters")

# List all spaces recursively
for space in project.get_spaces(recursive=True):
    print(f"  {space.path}")

# Update project settings
project.set_public(True)  # Enable public access
project.update(description="Updated description")

# Soft delete (deprecate)
project.delete()

# Hard delete (permanent)
project.delete(force=True)

__init__(pb, client)

Initialize a Project from a protobuf response.

Parameters:

• pb (ProjectResponse) – The protobuf ProjectResponse message.

• client (_Client) – The client instance for making API calls.

Return type:

None

__repr__()

Return a string representation of the Project.

Return type:

str

create_bundle(bundle_name, parent_path=None, metadata=None)

Create a new bundle within this project.

Bundles are special items that aggregate other items. They provide a way to group related items together and maintain an audit trail of membership changes through revision history.

Parameters:

• bundle_name (str) – The name of the bundle. Must be unique within the parent space.

• parent_path (Optional[str]) – Optional parent path for the bundle. If not provided, creates the bundle at the project root (/{project_name}).

• metadata (Optional[Dict[str, str]]) – Optional key-value metadata for the bundle.

Returns:

The newly created Bundle object with kind “bundle”.

Return type:

Bundle

Raises:

grpc.RpcError – If the bundle name is already taken or if there is a connection error.

See also

Bundle: The Bundle class. create_bundle(): Create bundle in a space.

Example:

>>> project = kumiho.get_project("film-2024")
>>> # Create at project root
>>> bundle = project.create_bundle("release-bundle")
>>>
>>> # Create in specific space
>>> bundle = project.create_bundle(
...     "character-bundle",
...     parent_path="/film-2024/assets"
... )
>>>
>>> # Add items to the bundle
>>> hero = project.get_space("models").get_item("hero", "model")
>>> bundle.add_member(hero)

create_item(item_name, kind, parent_path=None, metadata=None)

Create a new item within this project.

Parameters:

• item_name (str) – The name of the item. Must be unique within the parent space combined with the kind.

• kind (str) – The type/kind of the item (e.g., “model”, “texture”).

• parent_path (Optional[str]) – Optional parent path for the item. If not provided, creates the item at the project root (/{project_name}).

• metadata (Optional[Dict[str, str]]) – Optional key-value metadata for the item.

Returns:

The newly created Item object.

Return type:

Item

Raises:

grpc.RpcError – If the item name/kind combination is already taken or if there is a connection error.

Example:

>>> project = kumiho.get_project("film-2024")
>>> # Create at project root
>>> item = project.create_item("hero", "model")
>>>
>>> # Create in specific space
>>> item = project.create_item(
...     "hero",
...     "texture",
...     parent_path="/film-2024/assets"
... )

create_space(name, parent_path=None)

Create a space within this project.

Parameters:

• name (str) – The name of the space to create.

• parent_path (Optional[str]) – Optional parent path. If not provided, creates the space at the project root (e.g., “/project-name”).

Returns:

The newly created Space object.

Return type:

Space

Example

>>> project = kumiho.get_project("film-2024")
>>> # Create at root
>>> chars = project.create_space("characters")
>>> # Create nested space
>>> heroes = project.create_space("heroes", parent_path="/film-2024/characters")

delete(force=False)

Delete or deprecate this project.

Parameters:

force (bool) – If True, permanently delete the project and all its contents. If False (default), mark as deprecated.

Returns:

Response indicating success or failure.

Return type:

StatusResponse

Warning

Force deletion is irreversible and removes all spaces, items, revisions, artifacts, and edges within the project.

Example

>>> project = kumiho.get_project("old-project")
>>> # Soft delete (can be recovered)
>>> project.delete()
>>> # Hard delete (permanent)
>>> project.delete(force=True)

get_bundle(bundle_name, parent_path=None)

Get an existing bundle within this project.

Parameters:

• bundle_name (str) – The name of the bundle.

• parent_path (Optional[str]) – Optional parent path. If not provided, looks in the project root (/{project_name}).

Returns:

The Bundle object.

Return type:

Bundle

Raises:

• ValueError – If the item exists but is not a bundle.

• grpc.RpcError – If the bundle is not found.

Example:

>>> project = kumiho.get_project("film-2024")
>>> # Get from project root
>>> bundle = project.get_bundle("release-bundle")
>>>
>>> # Get from specific space
>>> bundle = project.get_bundle(
...     "character-bundle",
...     parent_path="/film-2024/assets"
... )
>>> members = bundle.get_members()

get_item(item_name, kind, parent_path=None)

Get an existing item within this project.

Parameters:

• item_name (str) – The name of the item.

• kind (str) – The type/kind of the item (e.g., “model”, “texture”).

• parent_path (Optional[str]) – Optional parent path. If not provided, looks in the project root (/{project_name}).

Returns:

The Item object.

Return type:

Item

Raises:

grpc.RpcError – If the item is not found.

Example:

>>> project = kumiho.get_project("film-2024")
>>> # Get from project root
>>> item = project.get_item("hero", "model")
>>>
>>> # Get from specific space
>>> item = project.get_item(
...     "hero",
...     "texture",
...     parent_path="/film-2024/assets"
... )

get_space(name, parent_path=None)

Get an existing space within this project.

Parameters:

• name (str) – The name of the space, or an absolute path starting with “/”.

• parent_path (Optional[str]) – Optional parent path if name is a relative name.

Returns:

The Space object.

Return type:

Space

Raises:

grpc.RpcError – If the space is not found.

Example

>>> # Get by absolute path
>>> space = project.get_space("/film-2024/characters")

>>> # Get by relative name (from project root)
>>> space = project.get_space("characters")

>>> # Get nested space with parent path
>>> heroes = project.get_space("heroes", parent_path="/film-2024/characters")

get_spaces(parent_path=None, recursive=False)

List spaces within this project.

Parameters:

• parent_path (Optional[str]) – Optional path to start from. Defaults to project root.

• recursive (bool) – If True, include all nested spaces. If False (default), only direct children.

Returns:

A list of Space objects.

Return type:

List[Space]

Example

>>> # List direct children only
>>> spaces = project.get_spaces()
>>> for s in spaces:
...     print(s.name)

>>> # List all spaces recursively
>>> all_spaces = project.get_spaces(recursive=True)
>>> for s in all_spaces:
...     print(s.path)

set_public(public)

Set whether this project allows anonymous read access.

Parameters:

public (bool) – True to enable public access, False to require authentication for all access.

Returns:

The updated Project object.

Return type:

Project

Example

>>> project.set_public(True)  # Enable public access
>>> project.set_public(False)  # Require authentication

update(description=None, allow_public=None)

Update project properties.

Parameters:

• description (Optional[str]) – New description for the project.

• allow_public (Optional[bool]) – New public access setting.

Returns:

The updated Project object.

Return type:

Project

Example

>>> project.update(
...     description="Updated project description",
...     allow_public=True
... )

class kumiho.Space

Bases: KumihoObject

A hierarchical container for organizing items in Kumiho.

Spaces form the folder structure within a project. They can contain other spaces (subspaces) and items, allowing you to organize assets in a meaningful hierarchy.

Spaces are identified by their full path (e.g., “/project/characters/heroes”) and can store custom metadata.

path

The full path of the space (e.g., “/project/assets”).

Type:

str

name

The name of this space (last component of path).

Type:

str

type

The type of space (“root” for project-level, “sub” for nested).

Type:

str

created_at

ISO timestamp when the space was created.

Type:

Optional[str]

author

The user ID who created the space.

Type:

str

metadata

Custom metadata key-value pairs.

Type:

Dict[str, str]

username

Display name of the creator.

Type:

str

Example

Creating and navigating spaces:

import kumiho

project = kumiho.get_project("my-project")

# Create a space
assets = project.create_space("assets")

# Create subspaces
models = assets.create_space("models")
textures = assets.create_space("textures")

# Create items
chair = models.create_item("chair", "model")

# List items with filters
all_models = models.get_items()
wood_textures = textures.get_items(name_filter="wood*")

# Navigate hierarchy
parent = models.get_parent_space()  # Returns assets
project = models.get_project()  # Returns my-project

__init__(pb_space, client)

Initialize a Space from a protobuf response.

Parameters:

• pb_space (SpaceResponse) – The protobuf SpaceResponse message.

• client (_Client) – The client instance for making API calls.

Return type:

None

__repr__()

Return a string representation of the Space.

Return type:

str

create_bundle(bundle_name, metadata=None)

Create a new bundle within this space.

Bundles are special items that aggregate other items. They provide a way to group related items together and maintain an audit trail of membership changes through revision history.

Parameters:

• bundle_name (str) – The name of the bundle. Must be unique within the space.

• metadata (Optional[Dict[str, str]]) – Optional key-value metadata for the bundle.

Returns:

The newly created Bundle object with kind “bundle”.

Return type:

Bundle

Raises:

grpc.RpcError – If the bundle name is already taken or if there is a connection error.

See also

Bundle: The Bundle class. create_bundle(): Create bundle in a project.

Example:

>>> # Create a bundle for a character bundle
>>> assets = project.get_space("assets")
>>> bundle = assets.create_bundle("character-bundle")
>>>
>>> # Add items to the bundle
>>> hero = assets.get_item("hero", "model")
>>> bundle.add_member(hero)

create_item(item_name, kind)

Create a new item within this space.

Items are versioned assets that can contain multiple artifacts. The combination of name and kind must be unique within the space.

Parameters:

• item_name (str) – The name of the item (e.g., “hero-character”).

• kind (str) – The kind of item (e.g., “model”, “texture”, “workflow”).

Returns:

The newly created Item object.

Return type:

Item

Example

>>> models = project.get_space("models")
>>> chair = models.create_item("office-chair", "model")
>>> wood = textures.create_item("oak-wood", "texture")

create_space(name)

Create a new subspace within this space.

Parameters:

name (str) – The name of the new subspace.

Returns:

The newly created Space object.

Return type:

Space

Example

>>> assets = project.create_space("assets")
>>> models = assets.create_space("models")
>>> textures = assets.create_space("textures")

delete(force=False)

Delete this space.

Parameters:

force (bool) – If True, force deletion even if the space contains items. If False (default), deletion fails if space is not empty.

Raises:

grpc.RpcError – If deletion fails (e.g., space not empty and force=False).

Return type:

None

Example

>>> # Delete empty space
>>> empty_space.delete()

>>> # Force delete space with contents
>>> old_space.delete(force=True)

delete_attribute(key)

Delete a single metadata attribute.

Parameters:

key (str) – The attribute key to delete.

Returns:

True if the attribute was deleted successfully.

Return type:

bool

Example

>>> space.delete_attribute("old_field")
True

get_attribute(key)

Get a single metadata attribute.

Parameters:

key (str) – The attribute key to retrieve.

Return type:

Optional[str]

Returns:

The attribute value if it exists, None otherwise.

Example

>>> space.get_attribute("department")
"modeling"

get_bundle(bundle_name)

Get a bundle by name from this space.

This is a convenience method that fetches a bundle item and returns it as a Bundle object with bundle-specific methods like add_member(), get_members(), etc.

Parameters:

bundle_name (str) – The name of the bundle.

Returns:

The Bundle object.

Return type:

Bundle

Raises:

grpc.RpcError – If the bundle is not found.

Example

>>> bundle = space.get_bundle("character-bundle")
>>> members = bundle.get_members()
>>> for member in members:
...     print(member.item_kref)

get_child_spaces()

Get immediate child spaces of this space.

This is a convenience method equivalent to get_spaces(recursive=False).

Returns:

A list of direct child Space objects.

Return type:

List[Space]

Example

>>> assets = project.get_space("assets")
>>> children = assets.get_child_spaces()
>>> for child in children:
...     print(child.name)

get_item(item_name, kind)

Get a specific item by name and kind.

Parameters:

• item_name (str) – The name of the item.

• kind (str) – The kind of the item.

Returns:

The Item object.

Return type:

Item

Raises:

grpc.RpcError – If the item is not found.

Example

>>> chair = models.get_item("office-chair", "model")
>>> revisions = chair.get_revisions()

get_items(item_name_filter='', kind_filter='')

List items within this space with optional filtering.

Parameters:

• item_name_filter (str) – Filter by item name. Supports wildcards.

• kind_filter (str) – Filter by item kind.

Returns:

A list of Item objects matching the filters.

Return type:

List[Item]

Example

>>> # All items in space
>>> items = space.get_items()

>>> # Only models
>>> models = space.get_items(kind_filter="model")

>>> # Items starting with "hero"
>>> heroes = space.get_items(item_name_filter="hero*")

get_parent_space()

Get the parent space of this space.

Returns:

The parent Space object, or None if this is

a project-level root space.

Return type:

Optional[Space]

Example

>>> heroes = project.get_space("characters/heroes")
>>> chars = heroes.get_parent_space()  # Returns "characters" space
>>> root = chars.get_parent_space()  # Returns None (project root)

get_project()

Get the project that contains this space.

Returns:

The parent Project object.

Return type:

Project

Example

>>> space = kumiho.get_item("kref://my-project/assets/hero.model").get_space()
>>> project = space.get_project()
>>> print(project.name)
my-project

get_space(name)

Get an existing subspace by name.

Parameters:

name (str) – The name of the subspace (not full path).

Returns:

The Space object.

Return type:

Space

Raises:

grpc.RpcError – If the space is not found.

Example

>>> assets = project.get_space("assets")
>>> models = assets.get_space("models")

get_spaces(recursive=False)

List child spaces under this space.

Parameters:

recursive (bool) – If True, include all nested spaces. If False (default), only direct children.

Returns:

A list of Space objects.

Return type:

List[Space]

Example

>>> # Direct children only
>>> children = space.get_spaces()

>>> # All nested spaces
>>> all_spaces = space.get_spaces(recursive=True)

set_attribute(key, value)

Set a single metadata attribute.

This allows granular updates to metadata without replacing the entire metadata map.

Parameters:

• key (str) – The attribute key to set.

• value (str) – The attribute value.

Returns:

True if the attribute was set successfully.

Return type:

bool

Example

>>> space.set_attribute("department", "modeling")
True

set_metadata(metadata)

Set or update metadata for this space.

Metadata is a dictionary of string key-value pairs that can store any custom information about the space.

Parameters:

metadata (Dict[str, str]) – Dictionary of metadata to set. Existing keys are overwritten, new keys are added.

Returns:

The updated Space object.

Return type:

Space

Example

>>> space.set_metadata({
...     "department": "modeling",
...     "supervisor": "jane.doe",
...     "status": "active"
... })

class kumiho.Item

Bases: KumihoObject

A versioned asset in the Kumiho system.

Items represent assets that can have multiple revisions, such as 3D models, textures, workflows, or any other type of creative content. Each item belongs to a space and is identified by a combination of name and kind.

The item’s kref (Kumiho Reference) is a URI that uniquely identifies it: kref://project/space/item.kind

kref

The unique reference URI for this item.

Type:

Kref

name

The full name including kind (e.g., “hero.model”).

Type:

str

item_name

The base name of the item (e.g., “hero”).

Type:

str

kind

The kind of item (e.g., “model”, “texture”).

Type:

str

created_at

ISO timestamp when the item was created.

Type:

Optional[str]

author

The user ID who created the item.

Type:

str

metadata

Custom metadata key-value pairs.

Type:

Dict[str, str]

deprecated

Whether the item is deprecated.

Type:

bool

username

Display name of the creator.

Type:

str

Example

Basic item operations:

import kumiho

# Get item by kref
item = kumiho.get_item("kref://film/chars/hero.model")

# Create revisions
v1 = item.create_revision()
v2 = item.create_revision(metadata={"notes": "Updated mesh"})

# Get specific revision
v1 = item.get_revision(1)
latest = item.get_latest_revision()

# Get revision by tag
approved = item.get_revision_by_tag("approved")

# Get revision at a specific time
historical = item.get_revision_by_time("202312011200")

# Set metadata
item.set_metadata({"status": "final", "priority": "high"})

# Deprecate the item
item.set_deprecated(True)

__init__(pb_item, client)

Initialize an Item from a protobuf response.

Parameters:

• pb_item (ItemResponse) – The protobuf ItemResponse message.

• client (_Client) – The client instance for making API calls.

Return type:

None

__repr__()

Return a string representation of the Item.

Return type:

str

create_revision(metadata=None, number=0)

Create a new revision of this item.

Revisions are automatically numbered sequentially. Each revision starts with the “latest” tag, which moves to the newest revision.

Parameters:

• metadata (Optional[Dict[str, str]]) – Optional metadata for the revision (e.g., artist notes, render settings, software versions).

• number (int) – Specific revision number to use. If 0 (default), auto-assigns the next available number.

Returns:

The newly created Revision object.

Return type:

Revision

Example

>>> # Auto-numbered revision
>>> v1 = item.create_revision()
>>> v2 = item.create_revision(metadata={"artist": "jane"})

>>> # Specific revision number (use with caution)
>>> v5 = item.create_revision(number=5)

delete(force=False)

Delete this item.

Parameters:

force (bool) – If True, permanently delete the item and all its revisions. If False (default), deletion may fail if the item has revisions.

Raises:

grpc.RpcError – If deletion fails.

Return type:

None

Example

>>> # Delete item (fails if has revisions)
>>> item.delete()

>>> # Force delete with all revisions
>>> item.delete(force=True)

delete_attribute(key)

Delete a single metadata attribute.

Parameters:

key (str) – The attribute key to delete.

Returns:

True if the attribute was deleted successfully.

Return type:

bool

Example

>>> item.delete_attribute("old_field")
True

get_attribute(key)

Get a single metadata attribute.

Parameters:

key (str) – The attribute key to retrieve.

Return type:

Optional[str]

Returns:

The attribute value if it exists, None otherwise.

Example

>>> item.get_attribute("status")
"final"

get_latest_revision()

Get the latest revision of this item.

The latest revision is the one with the “latest” tag, or if none exists, the revision with the highest number.

Returns:

The latest Revision object, or None if no

revisions exist.

Return type:

Optional[Revision]

Example

>>> latest = item.get_latest_revision()
>>> if latest:
...     print(f"Latest: v{latest.number}")

get_project()

Get the project that contains this item.

Returns:

The parent Project object.

Return type:

Project

Example

>>> project = item.get_project()
>>> print(project.name)

get_revision(revision_number)

Get a specific revision by its number.

Parameters:

revision_number (int) – The revision number to retrieve (1-based).

Returns:

The Revision object if found, None otherwise.

Return type:

Optional[Revision]

Example

>>> v3 = item.get_revision(3)
>>> if v3:
...     artifacts = v3.get_artifacts()

get_revision_by_tag(tag)

Get a revision by its tag.

Common tags include “latest”, “published”, “approved”, etc. Custom tags can be applied to revisions using Revision.tag().

Parameters:

tag (str) – The tag to search for.

Returns:

The Revision object if found, None otherwise.

Return type:

Optional[Revision]

Example

>>> approved = item.get_revision_by_tag("approved")
>>> published = item.get_revision_by_tag("published")

get_revision_by_time(time, tag=None)

Get the revision that had a specific tag at a given time.

This is essential for reproducible builds and historical queries. By combining a tag (like “published”) with a timestamp, you can answer questions like “What was the published version on June 1st?”

Parameters:

• time (Union[str, datetime]) – The time as a datetime object, or a string in either YYYYMMDDHHMM format (e.g., “202312251430”) or RFC3339 format (e.g., “2023-12-25T14:30:00Z”).

• tag (Optional[str]) – Optional tag to filter by. Common values: - “published”: Find the published revision at that time - “approved”: Find the approved revision at that time - None (default): Find the latest revision at that time

Returns:

The Revision that had the specified tag

at that time, or None if not found.

Return type:

Optional[Revision]

Example

>>> from datetime import datetime, timezone

>>> # Get the published revision as of June 1st, 2024
>>> june_1 = datetime(2024, 6, 1, tzinfo=timezone.utc)
>>> rev = item.get_revision_by_time(june_1, tag="published")

>>> # Get whatever was latest at a specific time
>>> rev = item.get_revision_by_time("202312251430")

>>> # Using RFC3339 format with published tag
>>> rev = item.get_revision_by_time(
...     "2024-06-01T00:00:00Z",
...     tag="published"
... )

Note

This is particularly useful with the “published” tag since published revisions are immutable and represent stable, approved versions suitable for downstream consumption.

get_revisions()

Get all revisions of this item.

Returns:

A list of Revision objects, ordered by revision number.

Return type:

List[Revision]

Example

>>> revisions = item.get_revisions()
>>> for v in revisions:
...     print(f"v{v.number}: created {v.created_at}")

get_space()

Get the space that contains this item.

Returns:

The parent Space object.

Return type:

Space

Example

>>> item = kumiho.get_item("kref://project/chars/hero.model")
>>> space = item.get_space()
>>> print(space.path)  # "/project/chars"

peek_next_revision()

Get the next revision number that would be assigned.

This is useful for previewing revision numbers before creating revisions, such as for naming files or planning workflows.

Returns:

The next revision number.

Return type:

int

Example

>>> next_num = item.peek_next_revision()
>>> print(f"Next revision will be v{next_num}")

set_attribute(key, value)

Set a single metadata attribute.

This allows granular updates to metadata without replacing the entire metadata map.

Parameters:

• key (str) – The attribute key to set.

• value (str) – The attribute value.

Returns:

True if the attribute was set successfully.

Return type:

bool

Example

>>> item.set_attribute("status", "final")
True

set_deprecated(status)

Set the deprecated status of this item.

Deprecated items are hidden from default searches but remain accessible for historical reference.

Parameters:

status (bool) – True to deprecate, False to restore.

Return type:

None

Example

>>> item.set_deprecated(True)  # Hide from searches
>>> item.set_deprecated(False)  # Restore visibility

set_metadata(metadata)

Set or update metadata for this item.

Metadata is merged with existing metadata—existing keys are overwritten and new keys are added.

Parameters:

metadata (Dict[str, str]) – Dictionary of metadata key-value pairs.

Returns:

The updated Item object.

Return type:

Item

Example

>>> item.set_metadata({
...     "status": "final",
...     "department": "modeling",
...     "complexity": "high"
... })

class kumiho.Revision

Bases: KumihoObject

A specific iteration of an item in the Kumiho system.

Revisions are immutable snapshots of an item at a point in time. Each revision can have multiple artifacts (file references), tags for categorization, and edges to other revisions for dependency tracking.

The revision’s kref includes the revision number: kref://project/space/item.kind?r=1

Revisions support dynamic tag checking—the tags property automatically refreshes from the server if the local data might be stale (older than 5 seconds). This ensures tags like “latest” are always current.

kref

The unique reference URI for this revision.

Type:

Kref

item_kref

Reference to the parent item.

Type:

Kref

number

The revision number (1-based).

Type:

int

latest

Whether this is currently the latest revision.

Type:

bool

tags

Tags applied to this revision (auto-refreshes).

Type:

List[str]

metadata

Custom metadata key-value pairs.

Type:

Dict[str, str]

created_at

ISO timestamp when the revision was created.

Type:

Optional[str]

author

The user ID who created the revision.

Type:

str

deprecated

Whether the revision is deprecated.

Type:

bool

published

Whether the revision is published.

Type:

bool

username

Display name of the creator.

Type:

str

default_artifact

Name of the default artifact.

Type:

Optional[str]

Example

Creating and managing revisions:

import kumiho

item = kumiho.get_item("kref://project/models/hero.model")

# Create a revision with metadata
v1 = item.create_revision(metadata={
    "artist": "jane.doe",
    "software": "maya-2024",
    "notes": "Initial model"
})

# Add artifacts
mesh = v1.create_artifact("mesh", "/assets/hero.fbx")
rig = v1.create_artifact("rig", "/assets/hero_rig.fbx")

# Set default artifact (for resolve)
v1.set_default_artifact("mesh")

# Tag the revision
v1.tag("approved")

# Check tags
if v1.has_tag("approved"):
    print("Revision is approved!")

# Get all artifacts
for r in v1.get_artifacts():
    print(f"  {r.name}: {r.location}")

# Edge to dependencies
texture = kumiho.get_revision("kref://project/tex/skin.texture?r=2")
v1.create_edge(texture, kumiho.DEPENDS_ON)

__init__(pb_revision, client)

Initialize a Revision from a protobuf response.

Parameters:

• pb_revision (RevisionResponse) – The protobuf RevisionResponse message.

• client (_Client) – The client instance for making API calls.

Return type:

None

__repr__()

Return a string representation of the Revision.

Return type:

str

analyze_impact(edge_type_filter=None, max_depth=10, limit=100)

Analyze the impact of changes to this revision.

Returns all revisions that directly or indirectly depend on this revision, sorted by impact depth (closest dependencies first).

Parameters:

• edge_type_filter (Optional[List[str]]) – Edge types to follow (default: all).

• max_depth (int) – How far to traverse (default: 10).

• limit (int) – Maximum results (default: 100).

Returns:

Revisions that would be impacted.

Return type:

List[ImpactedRevision]

Example

>>> # Before modifying a shared texture
>>> impact = texture_v1.analyze_impact()
>>> print(f"{len(impact)} revisions would need review")
>>> for iv in impact[:5]:
...     print(f"  {iv.revision_kref} (depth {iv.impact_depth})")

create_artifact(name, location)

Create a new artifact for this revision.

Artifacts are file references that point to actual assets on disk or network storage. Kumiho tracks the path and metadata but does not upload or copy the files.

Parameters:

• name (str) – The name of the artifact (e.g., “mesh”, “textures”, “rig”).

• location (str) – The file path or URI where the artifact is stored.

Returns:

The newly created Artifact object.

Return type:

Artifact

Example

>>> mesh = revision.create_artifact("mesh", "/assets/hero.fbx")
>>> textures = revision.create_artifact("textures", "smb://server/tex/hero.zip")

create_edge(target_revision, edge_type, metadata=None)

Create an edge from this revision to another revision.

Edges represent relationships between revisions, such as dependencies, references, or derivations. This is useful for tracking asset lineage.

Parameters:

• target_revision (Revision) – The target revision to link to.

• edge_type (str) – The type of edge. Use constants from kumiho.EdgeType: - kumiho.DEPENDS_ON: This revision depends on target. - kumiho.DERIVED_FROM: This revision was derived from target. - kumiho.REFERENCED: This revision references target. - kumiho.CONTAINS: This revision contains target.

• metadata (Optional[Dict[str, str]]) – Optional metadata for the edge.

Returns:

The created Edge object.

Return type:

Edge

Example

>>> import kumiho

>>> # Edge to a texture dependency
>>> texture = kumiho.get_revision("kref://project/tex/skin.texture?r=2")
>>> revision.create_edge(texture, kumiho.DEPENDS_ON)

>>> # Edge with metadata
>>> base = kumiho.get_revision("kref://project/models/base.model?r=1")
>>> revision.create_edge(base, kumiho.DERIVED_FROM, {
...     "modification": "Added details"
... })

delete(force=False)

Delete this revision.

Parameters:

force (bool) – If True, force deletion even if the revision has artifacts. If False (default), deletion may fail.

Raises:

grpc.RpcError – If deletion fails.

Return type:

None

Example

>>> revision.delete()  # Fails if has artifacts
>>> revision.delete(force=True)  # Force delete

delete_attribute(key)

Delete a single metadata attribute.

Parameters:

key (str) – The attribute key to delete.

Returns:

True if the attribute was deleted successfully.

Return type:

bool

Example

>>> revision.delete_attribute("old_field")
True

delete_edge(target_revision, edge_type)

Delete an edge from this revision.

Parameters:

• target_revision (Revision) – The target revision of the edge.

• edge_type (str) – The type of edge to delete.

Return type:

None

Example

>>> revision.delete_edge(texture_revision, kumiho.DEPENDS_ON)

find_path_to(target, edge_type_filter=None, max_depth=10, all_paths=False)

Find the shortest path from this revision to another.

Uses graph traversal to find how two revisions are connected.

Parameters:

• target (Revision) – The target revision to find a path to.

• edge_type_filter (Optional[List[str]]) – Filter by edge types.

• max_depth (int) – Maximum path length to search.

• all_paths (bool) – If True, returns all shortest paths via result.paths.

Return type:

Optional[RevisionPath]

Returns:

RevisionPath if a path exists, None otherwise. Use all_paths=True and access result.paths for multiple paths.

Example

>>> path = model_v1.find_path_to(texture_v3)
>>> if path:
...     print(f"Path length: {path.total_depth}")
...     for step in path.steps:
...         print(f"  -> {step.revision_kref} via {step.edge_type}")

get_all_dependencies(edge_type_filter=None, max_depth=10, limit=100)

Get all transitive dependencies of this revision.

Traverses outgoing edges to find all revisions this revision depends on, directly or indirectly.

Parameters:

• edge_type_filter (Optional[List[str]]) – Filter by edge types (e.g., [kumiho.DEPENDS_ON]).

• max_depth (int) – Maximum traversal depth (default: 10, max: 20).

• limit (int) – Maximum number of results (default: 100, max: 1000).

Returns:

Contains all discovered revisions and paths.

Return type:

TraversalResult

Example

>>> import kumiho

>>> # Get all dependencies up to 5 hops
>>> deps = revision.get_all_dependencies(
...     edge_type_filter=[kumiho.DEPENDS_ON],
...     max_depth=5
... )
>>> for kref in deps.revision_krefs:
...     print(f"Depends on: {kref}")

get_all_dependents(edge_type_filter=None, max_depth=10, limit=100)

Get all revisions that transitively depend on this revision.

Traverses incoming edges to find all revisions that depend on this revision, directly or indirectly. Useful for impact analysis.

Parameters:

• edge_type_filter (Optional[List[str]]) – Filter by edge types.

• max_depth (int) – Maximum traversal depth.

• limit (int) – Maximum number of results.

Returns:

Contains all dependent revisions.

Return type:

TraversalResult

Example

>>> # Find everything that would be affected by changing this texture
>>> dependents = texture_v1.get_all_dependents([kumiho.DEPENDS_ON])
>>> print(f"{len(dependents.revision_krefs)} revisions would be affected")

get_artifact(name)

Get a specific artifact by name from this revision.

Parameters:

name (str) – The name of the artifact.

Returns:

The Artifact object.

Return type:

Artifact

Raises:

grpc.RpcError – If the artifact is not found.

Example

>>> mesh = revision.get_artifact("mesh")
>>> print(mesh.location)

get_artifacts()

Get all artifacts associated with this revision.

Returns:

A list of Artifact objects.

Return type:

List[Artifact]

Example

>>> for artifact in revision.get_artifacts():
...     print(f"{artifact.name}: {artifact.location}")

get_attribute(key)

Get a single metadata attribute.

Parameters:

key (str) – The attribute key to retrieve.

Return type:

Optional[str]

Returns:

The attribute value if it exists, None otherwise.

Example

>>> revision.get_attribute("render_engine")
"cycles"

get_edges(edge_type_filter=None, direction=0)

Get edges involving this revision.

Parameters:

• edge_type_filter (Optional[str]) – Optional filter for edge type.

• direction (int) – The direction of edges to retrieve: - kumiho.OUTGOING (0): Edges from this revision. - kumiho.INCOMING (1): Edges to this revision. - kumiho.BOTH (2): Edges in both directions.

Returns:

A list of Edge objects.

Return type:

List[Edge]

Example

>>> import kumiho

>>> # Get all dependencies
>>> deps = revision.get_edges(kumiho.DEPENDS_ON, kumiho.OUTGOING)

>>> # Get all revisions that depend on this one
>>> dependents = revision.get_edges(kumiho.DEPENDS_ON, kumiho.INCOMING)

get_item()

Get the parent item of this revision.

Returns:

The Item object that contains this revision.

Return type:

Item

Example

>>> item = revision.get_item()
>>> print(item.item_name)

get_locations()

Get the file locations of all artifacts in this revision.

This is a convenience method to quickly get all file paths.

Returns:

A list of file location strings.

Return type:

List[str]

Example

>>> locations = revision.get_locations()
>>> for loc in locations:
...     print(loc)

get_project()

Get the project that contains this revision.

Returns:

The Project object.

Return type:

Project

Example

>>> project = revision.get_project()
>>> print(project.name)

get_space()

Get the space that contains this revision’s item.

Returns:

The Space object.

Return type:

Space

Example

>>> space = revision.get_space()
>>> print(space.path)

has_tag(tag)

Check if this revision currently has a specific tag.

This makes a server call to ensure the tag status is current.

Parameters:

tag (str) – The tag to check for.

Returns:

True if the revision has the tag, False otherwise.

Return type:

bool

Example

>>> if revision.has_tag("approved"):
...     print("Ready for production!")

refresh()

Refresh this revision’s data from the server.

This updates all properties to reflect the current state in the database, including tags that may have changed (like “latest”).

Example

>>> revision.refresh()
>>> print(revision.tags)  # Now shows current tags

Return type:

None

set_attribute(key, value)

Set a single metadata attribute.

This allows granular updates to metadata without replacing the entire metadata map.

Parameters:

• key (str) – The attribute key to set.

• value (str) – The attribute value.

Returns:

True if the attribute was set successfully.

Return type:

bool

Example

>>> revision.set_attribute("render_engine", "cycles")
True

set_default_artifact(artifact_name)

Set the default artifact for this revision.

The default artifact is used when resolving the revision’s kref without specifying an artifact name.

Parameters:

artifact_name (str) – The name of the artifact to set as default.

Return type:

None

Example

>>> revision.set_default_artifact("mesh")
>>> # Now kref://project/model.kind?r=1 resolves to the mesh

set_deprecated(status)

Set the deprecated status of this revision.

Deprecated revisions are hidden from default queries but remain accessible for historical reference.

Parameters:

status (bool) – True to deprecate, False to restore.

Return type:

None

Example

>>> revision.set_deprecated(True)  # Hide from queries

set_metadata(metadata)

Set or update metadata for this revision.

Metadata is merged with existing metadata—existing keys are overwritten and new keys are added.

Parameters:

metadata (Dict[str, str]) – Dictionary of metadata key-value pairs.

Returns:

The updated Revision object.

Return type:

Revision

Example

>>> revision.set_metadata({
...     "render_engine": "arnold",
...     "frame_range": "1-100",
...     "resolution": "4K"
... })

tag(tag)

Apply a tag to this revision.

Tags are used to categorize revisions and mark their status. Common tags include “latest”, “published”, “approved”, etc.

Note

The “latest” tag is automatically managed—it always points to the newest revision.

Parameters:

tag (str) – The tag to apply.

Return type:

None

Example

>>> revision.tag("approved")
>>> revision.tag("ready-for-lighting")

property tags: List[str]

Get the current tags for this revision.

This property automatically refreshes from the server if the data might be stale (older than 5 seconds), ensuring dynamic tags like “latest” are always current.

Returns:

The list of tags on this revision.

Return type:

List[str]

Example

>>> revision = item.get_revision(1)
>>> print(revision.tags)  # ['latest', 'approved']

untag(tag)

Remove a tag from this revision.

Parameters:

tag (str) – The tag to remove.

Return type:

None

Example

>>> revision.untag("work-in-progress")

was_tagged(tag)

Check if this revision was ever tagged with a specific tag.

This checks the historical record, not just current tags.

Parameters:

tag (str) – The tag to check for.

Returns:

True if the revision was ever tagged with this tag.

Return type:

bool

Example

>>> if revision.was_tagged("approved"):
...     print("Was approved at some point")

class kumiho.Artifact

Bases: KumihoObject

A file reference within a revision in the Kumiho system.

Artifacts are the leaf nodes of the Kumiho hierarchy. They point to actual files on local disk, network storage, or cloud URIs. Kumiho tracks the path and metadata but does not upload or modify the files.

The artifact’s kref includes both revision and artifact name: kref://project/space/item.kind?r=1&a=artifact_name

kref

The unique reference URI for this artifact.

Type:

Kref

location

The file path or URI where the artifact is stored.

Type:

str

revision_kref

Reference to the parent revision.

Type:

Kref

item_kref

Reference to the parent item.

Type:

Optional[Kref]

created_at

ISO timestamp when the artifact was created.

Type:

Optional[str]

author

The user ID who created the artifact.

Type:

str

metadata

Custom metadata key-value pairs.

Type:

Dict[str, str]

deprecated

Whether the artifact is deprecated.

Type:

bool

username

Display name of the creator.

Type:

str

Example

Working with artifacts:

import kumiho

revision = kumiho.get_revision("kref://project/models/hero.model?r=1")

# Create artifacts
mesh = revision.create_artifact("mesh", "/assets/hero.fbx")
rig = revision.create_artifact("rig", "/assets/hero_rig.fbx")
textures = revision.create_artifact("textures", "smb://server/tex/hero/")

# Set metadata
mesh.set_metadata({
    "triangles": "2.5M",
    "format": "FBX 2020",
    "units": "centimeters"
})

# Set as default artifact
mesh.set_default()

# Get artifact by name
retrieved = revision.get_artifact("mesh")
print(f"Location: {retrieved.location}")

# Navigate hierarchy
item = mesh.get_item()
project = mesh.get_project()

__init__(pb_artifact, client)

Initialize an Artifact from a protobuf response.

Parameters:

• pb_artifact (ArtifactResponse) – The protobuf ArtifactResponse message.

• client (_Client) – The client instance for making API calls.

Return type:

None

__repr__()

Return a string representation of the Artifact.

Return type:

str

delete(force=False)

Delete this artifact.

Parameters:

force (bool) – If True, force deletion. If False (default), normal deletion rules apply.

Raises:

grpc.RpcError – If deletion fails.

Return type:

None

Example

>>> artifact.delete()

delete_attribute(key)

Delete a single metadata attribute.

Parameters:

key (str) – The attribute key to delete.

Returns:

True if the attribute was deleted successfully.

Return type:

bool

Example

>>> artifact.delete_attribute("old_field")
True

get_attribute(key)

Get a single metadata attribute.

Parameters:

key (str) – The attribute key to retrieve.

Return type:

Optional[str]

Returns:

The attribute value if it exists, None otherwise.

Example

>>> artifact.get_attribute("file_size")
"125MB"

get_item()

Get the item that contains this artifact.

Returns:

The Item object.

Return type:

Item

Example

>>> item = artifact.get_item()
>>> print(item.item_name)

get_project()

Get the project containing this artifact.

Returns:

The Project object.

Return type:

Project

Example

>>> project = artifact.get_project()
>>> print(project.name)

get_revision()

Get the parent revision of this artifact.

Returns:

The Revision object that contains this artifact.

Return type:

Revision

Example

>>> revision = artifact.get_revision()
>>> print(f"Revision {revision.number}")

get_space()

Get the space containing this artifact’s item.

Returns:

The Space object.

Return type:

Space

Example

>>> space = artifact.get_space()
>>> print(space.path)

property name: str

Get the artifact name from its kref.

Returns:

The artifact name extracted from the kref URI.

Return type:

str

Example

>>> artifact = revision.get_artifact("mesh")
>>> print(artifact.name)  # "mesh"

set_attribute(key, value)

Set a single metadata attribute.

This allows granular updates to metadata without replacing the entire metadata map.

Parameters:

• key (str) – The attribute key to set.

• value (str) – The attribute value.

Returns:

True if the attribute was set successfully.

Return type:

bool

Example

>>> artifact.set_attribute("file_size", "125MB")
True

set_default()

Set this artifact as the default for its revision.

The default artifact is used when resolving the revision’s kref without specifying an artifact name.

Example

>>> mesh = revision.create_artifact("mesh", "/assets/model.fbx")
>>> mesh.set_default()
>>> # Now resolving the revision kref returns this artifact's location

Return type:

None

set_deprecated(status)

Set the deprecated status of this artifact.

Deprecated artifacts are hidden from default queries but remain accessible for historical reference.

Parameters:

status (bool) – True to deprecate, False to restore.

Return type:

None

Example

>>> artifact.set_deprecated(True)  # Hide from queries
>>> artifact.set_deprecated(False)  # Restore visibility

set_metadata(metadata)

Set or update metadata for this artifact.

Metadata is merged with existing metadata—existing keys are overwritten and new keys are added.

Parameters:

metadata (Dict[str, str]) – Dictionary of metadata key-value pairs.

Returns:

The updated Artifact object.

Return type:

Artifact

Example

>>> artifact.set_metadata({
...     "file_size": "125MB",
...     "format": "FBX 2020",
...     "triangles": "2.5M",
...     "software": "Maya 2024"
... })

class kumiho.Edge

Bases: KumihoObject

A relationship between two revisions in the Kumiho system.

Edges represent semantic relationships between revisions, enabling dependency tracking, lineage visualization, and impact analysis. They are directional (source -> target) and typed.

Common use cases:

• Track which textures a model uses (DEPENDS_ON)

• Record that a LOD was created from a high-poly model (DERIVED_FROM)

• Link a render to the scene file that created it (CREATED_FROM)

source_kref

Reference to the source revision.

Type:

Kref

target_kref

Reference to the target revision.

Type:

Kref

edge_type

The type of relationship (see EdgeType).

Type:

str

metadata

Custom metadata key-value pairs.

Type:

Dict[str, str]

created_at

ISO timestamp when the edge was created.

Type:

Optional[str]

author

The user ID who created the edge.

Type:

str

username

Display name of the creator.

Type:

str

Example:

import kumiho

# Get revisions
model = kumiho.get_revision("kref://project/models/hero.model?r=1")
texture = kumiho.get_revision("kref://project/tex/skin.texture?r=2")

# Create edge with metadata
edge = model.create_edge(texture, kumiho.DEPENDS_ON, {
    "channel": "diffuse",
    "uv_set": "0"
})

# Inspect edge
print(f"Type: {edge.edge_type}")
print(f"From: {edge.source_kref}")
print(f"To: {edge.target_kref}")

# Delete edge
edge.delete()

__init__(pb_edge, client)

Initialize an Edge from a protobuf message.

Parameters:

• pb_edge (Edge) – The protobuf Edge message.

• client (_Client) – The client instance for making API calls.

Return type:

None

__repr__()

Return a string representation of the Edge.

Return type:

str

delete(force=False)

Delete this edge.

Parameters:

force (bool) – Reserved for future use.

Return type:

None

Example

>>> edge = model.create_edge(texture, kumiho.DEPENDS_ON)
>>> edge.delete()  # Remove the relationship

class kumiho.Kref

Bases: str

A Kumiho Artifact Reference (URI-based unique identifier).

Kref is a subclass of str, so it behaves like a string but provides utility methods for parsing and extracting components from the URI.

The kref format is:

kref://project/space/item.kind?r=REVISION&a=ARTIFACT

uri

The URI string (for backward compatibility).

Type:

str

Example:

from kumiho import Kref

# Create from string
kref = Kref("kref://my-project/assets/hero.model?r=2")

# Use as string (since Kref extends str)
print(kref)  # kref://my-project/assets/hero.model?r=2

# Parse components
print(kref.get_space())    # "my-project/assets"
print(kref.get_revision()) # 2

# Compare with strings
if kref == "kref://my-project/assets/hero.model?r=2":
    print("Match!")

Note

Since Kref is a string subclass, you can use it anywhere a string is expected. All string methods work normally.

__eq__(other)

Compare with another Kref or string.

Return type:

bool

Parameters:

other (object)

__hash__()

Return hash for use in sets and dicts.

Return type:

int

static __new__(cls, uri, *, validate=True)

Create a new Kref instance.

Parameters:

• uri (str) – The kref URI string.

• validate (bool) – Whether to validate the URI (default: True). Set to False for trusted internal sources.

Returns:

A Kref instance that is also a string.

Return type:

Kref

Raises:

KrefValidationError – If validate=True and the URI is invalid.

Example

>>> kref = Kref("kref://project/space/item.kind?r=1")
>>> isinstance(kref, str)
True

__repr__()

Return a developer-friendly string representation.

Return type:

str

classmethod from_pb(pb_kref)

Create a Kref from a protobuf message.

This is used for krefs received from the server, which are trusted.

Parameters:

pb_kref (Kref) – The protobuf Kref message.

Returns:

A Kref instance.

Return type:

Kref

get_artifact_name()

Extract the artifact name from the URI query string.

Returns:

The artifact name if present, None otherwise.

Return type:

Optional[str]

Example

>>> Kref("kref://project/models/hero.model?r=1&a=mesh").get_artifact_name()
'mesh'
>>> Kref("kref://project/models/hero.model?r=1").get_artifact_name()
None

get_item_name()

Extract the item name with kind from the URI.

Returns:

The item name including kind (e.g., “hero.model”).

Return type:

str

Example

>>> Kref("kref://project/models/hero.model").get_item_name()
'hero.model'

get_kind()

Extract the item kind from the URI.

Returns:

The item kind (e.g., “model”, “texture”).

Return type:

str

Example

>>> Kref("kref://project/models/hero.model").get_kind()
'model'

get_path()

Extract the path component from the URI.

Returns the part after kref:// and before any query parameters.

Returns:

The path (e.g., “project/space/item.kind”).

Return type:

str

Example

>>> Kref("kref://project/models/hero.model?r=1").get_path()
'project/models/hero.model'

get_revision()

Extract the revision number from the URI query string.

Returns:

The revision number, or 1 if not specified.

Return type:

int

Example

>>> Kref("kref://project/models/hero.model?r=3").get_revision()
3
>>> Kref("kref://project/models/hero.model").get_revision()
1

get_space()

Extract the space path from the URI.

Returns the path up to but not including the item name.

Returns:

The space path (e.g., “project/models”).

Return type:

str

Example

>>> Kref("kref://project/models/hero.model").get_space()
'project/models'

to_pb()

Convert to a protobuf Kref object.

Used internally for gRPC communication.

Returns:

A protobuf Kref message.

Return type:

kumiho_pb2.Kref

property uri: str

Get the URI string.

This property exists for backward compatibility with older code that accessed .uri directly.

Returns:

The kref URI string.

Return type:

str

class kumiho.Event

Bases: object

A real-time notification from the Kumiho server.

Events represent changes or actions that occurred in the Kumiho system, such as creating versions, applying tags, or deleting resources. Use kumiho.event_stream() to subscribe to events.

routing_key

The event type identifier (e.g., “version.created”, “version.tagged”). Use wildcards in filters to match patterns.

Type:

str

kref

Reference to the affected object.

Type:

Kref

timestamp

ISO timestamp when the event occurred.

Type:

Optional[str]

author

The user ID who triggered the event.

Type:

str

details

Additional event-specific information (e.g., tag name for tagged events).

Type:

Dict[str, str]

cursor

Opaque cursor for resumable streaming. Save this value and pass it to event_stream(cursor=...) to resume from this event after reconnection. Only available on Creator tier and above.

Type:

Optional[str]

Example:

import kumiho

# React to revision creation with cursor tracking
last_cursor = None
try:
    for event in kumiho.event_stream(
        routing_key_filter="revision.created",
        cursor=last_cursor
    ):
        revision = kumiho.get_revision(event.kref)
        print(f"New revision: {revision.item_kref} v{revision.number}")
        print(f"  Created by: {event.author}")
        print(f"  At: {event.timestamp}")
        last_cursor = event.cursor  # Save cursor
except ConnectionError:
    # Reconnect using saved cursor
    pass

# Filter by kref pattern
for event in kumiho.event_stream(
    routing_key_filter="*",
    kref_filter="kref://production-project/**"
):
    print(f"Production change: {event.routing_key}")

__init__(pb_event)

Initialize an Event from a protobuf message.

Parameters:

pb_event (Event) – The protobuf Event message from the server.

Return type:

None

__repr__()

Return a string representation of the Event.

Return type:

str

exception kumiho.ProjectLimitError

Bases: Exception

Raised when guardrails block project creation (e.g., max projects reached).

class kumiho.Bundle

Bases: Item

A special item type that aggregates other items.

Bundles provide a way to group related items together. Unlike regular items, bundles cannot be created using the standard create_item method—the bundle kind is reserved.

Use Project.create_bundle() or Space.create_bundle() to create bundles.

Key features:

• Aggregates items (not revisions) via COLLECTS relationships.

• Each membership change creates a new revision for audit trail.

• Revision metadata is immutable, providing complete history.

• Cannot contain itself (self-reference protection).

kref

The unique identifier for this bundle.

Type:

Kref

name

The combined name (e.g., “my-bundle.bundle”).

Type:

str

item_name

The bundle name (e.g., “my-bundle”).

Type:

str

kind

Always “bundle”.

Type:

str

metadata

Custom metadata key-value pairs.

Type:

Dict[str, str]

created_at

ISO timestamp when the bundle was created.

Type:

str

author

The user ID who created the bundle.

Type:

str

username

Display name of the creator.

Type:

str

deprecated

Whether the bundle is deprecated.

Type:

bool

Example:

import kumiho

# Create a bundle from a project
project = kumiho.get_project("film-2024")
bundle = project.create_bundle("release-v1")

# Add items
model = kumiho.get_item("kref://film-2024/models/hero.model")
texture = kumiho.get_item("kref://film-2024/textures/hero.texture")
bundle.add_member(model)
bundle.add_member(texture)

# List current members
for member in bundle.get_members():
    print(f"{member.item_kref} added by {member.added_by_username}")

# Remove a member
bundle.remove_member(model)

# View complete audit history
for entry in bundle.get_history():
    print(f"v{entry.revision_number}: {entry.action}")

See also

Project.create_bundle(): Create a bundle in a project. Space.create_bundle(): Create a bundle in a space. BundleMember: Data class for member information. BundleRevisionHistory: Data class for audit entries.

__init__(pb, client)

Initialize a Bundle from a protobuf response.

Parameters:

• pb (ItemResponse) – The ItemResponse protobuf message.

• client (_Client) – The client instance for making subsequent calls.

Raises:

ValueError – If the kind is not ‘bundle’.

Return type:

None

__repr__()

Return a string representation of the Bundle.

Return type:

str

add_member(member, metadata=None)

Add an item to this bundle.

Creates a new revision of the bundle to track the change. The revision metadata will include the action ("ADDED") and the member item kref for audit purposes.

Parameters:

• member (Item) – The item to add to the bundle.

• metadata (Optional[Dict[str, str]]) – Optional additional metadata to store in the revision. This metadata becomes part of the immutable audit trail.

Returns:

A tuple containing:

• success: Whether the operation succeeded.

• message: A status message.

• new_revision: The new bundle revision created for this change.

Return type:

Tuple[bool, str, Optional[Revision]]

Raises:

• ValueError – If trying to add the bundle to itself.

• grpc.RpcError – If the member is already in the bundle (status code ALREADY_EXISTS).

Example:

hero_model = kumiho.get_item("kref://project/models/hero.model")

# Add with optional metadata
success, msg, revision = bundle.add_member(
    hero_model,
    metadata={"reason": "character bundle", "approved_by": "director"}
)

if success:
    print(f"Added in revision {revision.number}")

get_history()

Get the full history of membership changes.

Returns all revisions with their associated actions, providing a complete and immutable audit trail of all adds and removes.

The history is ordered by revision number, starting with the initial "CREATED" action.

Returns:

List of history entries, ordered

by revision number (oldest first).

Return type:

List[BundleRevisionHistory]

Example:

history = bundle.get_history()

for entry in history:
    print(f"Revision {entry.revision_number}:")
    print(f"  Action: {entry.action}")
    print(f"  By: {entry.username}")
    print(f"  At: {entry.created_at}")
    if entry.member_item_kref:
        print(f"  Item: {entry.member_item_kref}")

get_members(revision_number=None)

Get all items that are members of this bundle.

Returns information about each member item, including when it was added and by whom.

Parameters:

revision_number (Optional[int]) – Optional specific revision to query. If not provided, returns current membership.

Returns:

List of member information objects.

Return type:

List[BundleMember]

Example:

# Get current members
members = bundle.get_members()
for member in members:
    print(f"{member.item_kref}")
    print(f"  Added by: {member.added_by_username}")
    print(f"  In revision: {member.added_in_revision}")

# Get empty list if no members
if not members:
    print("Bundle is empty")

remove_member(member, metadata=None)

Remove an item from this bundle.

Creates a new revision of the bundle to track the change. The revision metadata will include the action ("REMOVED") and the member item kref for audit purposes.

Parameters:

• member (Item) – The item to remove from the bundle.

• metadata (Optional[Dict[str, str]]) – Optional additional metadata to store in the revision. This metadata becomes part of the immutable audit trail.

Returns:

A tuple containing:

• success: Whether the operation succeeded.

• message: A status message.

• new_revision: The new bundle revision created for this change.

Return type:

Tuple[bool, str, Optional[Revision]]

Raises:

grpc.RpcError – If the member is not in the bundle (status code NOT_FOUND).

Example:

# Remove an item from the bundle
success, msg, revision = bundle.remove_member(hero_model)

if success:
    print(f"Removed in revision {revision.number}")

class kumiho.BundleMember

Bases: object

An item that is a member of a bundle.

Represents the membership relationship between an item and a bundle, including metadata about when and by whom the item was added.

item_kref

The kref of the member item.

Type:

Kref

added_at

ISO timestamp when the item was added.

Type:

str

added_by

UUID of the user who added the item.

Type:

str

added_by_username

Display name of the user who added the item.

Type:

str

added_in_revision

The bundle revision when this item was added.

Type:

int

Example:

members = bundle.get_members()
for member in members:
    print(f"Item: {member.item_kref}")
    print(f"Added by: {member.added_by_username}")
    print(f"Added at: {member.added_at}")
    print(f"In revision: {member.added_in_revision}")

__init__(item_kref, added_at, added_by, added_by_username, added_in_revision)

Parameters:

• item_kref (Kref)

• added_at (str)

• added_by (str)

• added_by_username (str)

• added_in_revision (int)

Return type:

None

item_kref: Kref

The kref of the member item.

Type:

Kref

added_at: str

ISO timestamp when the item was added to the bundle.

Type:

str

added_by: str

UUID of the user who added the item.

Type:

str

added_by_username: str

Display name of the user who added the item.

Type:

str

added_in_revision: int

The bundle revision number when this item was added.

Type:

int

class kumiho.BundleRevisionHistory

Bases: object

A historical change to a bundle’s membership.

Each entry captures a single add or remove operation, providing an immutable audit trail of all membership changes. The metadata is immutable once created, ensuring complete traceability.

revision_number

The bundle revision number for this change.

Type:

int

action

The action performed: "CREATED", "ADDED", or "REMOVED".

Type:

str

member_item_kref

The item that was added/removed. None for the initial "CREATED" action.

Type:

Optional[Kref]

author

UUID of the user who made the change.

Type:

str

username

Display name of the user who made the change.

Type:

str

created_at

ISO timestamp of the change.

Type:

str

metadata

Immutable metadata captured at the time of change.

Type:

Dict[str, str]

Example:

history = bundle.get_history()
for entry in history:
    print(f"Revision {entry.revision_number}: {entry.action}")
    if entry.member_item_kref:
        print(f"  Item: {entry.member_item_kref}")
    print(f"  By: {entry.username} at {entry.created_at}")

__init__(revision_number, action, member_item_kref, author, username, created_at, metadata)

Parameters:

• revision_number (int)

• action (str)

• member_item_kref (Kref | None)

• author (str)

• username (str)

• created_at (str)

• metadata (Dict[str, str])

Return type:

None

revision_number: int

The bundle revision number for this change.

Type:

int

action: str

The action performed: "CREATED", "ADDED", or "REMOVED".

Type:

str

member_item_kref: Optional[Kref]

The item that was added/removed (None for CREATED).

Type:

Optional[Kref]

author: str

UUID of the user who made the change.

Type:

str

username: str

Display name of the user who made the change.

Type:

str

created_at: str

ISO timestamp of when the change was made.

Type:

str

metadata: Dict[str, str]

Immutable metadata captured at the time of the change.

Type:

Dict[str, str]

exception kumiho.ReservedKindError

Bases: Exception

Raised when attempting to create an item with a reserved kind.

This error is raised when calling Space.create_item() or the low-level client create_item with a reserved kind such as bundle.

Example:

import kumiho

space = project.get_space("assets")

# This will raise ReservedKindError
try:
    space.create_item("my-bundle", "bundle")
except kumiho.ReservedKindError as e:
    print(f"Error: {e}")
    # Use create_bundle instead
    bundle = space.create_bundle("my-bundle")

exception kumiho.KrefValidationError

Bases: ValueError

Raised when a Kref URI is invalid or contains malicious patterns.

exception kumiho.EdgeTypeValidationError

Bases: ValueError

Raised when an edge type is invalid or potentially malicious.

kumiho.validate_kref(uri)

Validate a Kref URI for security and correctness.

Checks for: - Proper kref:// scheme - No path traversal patterns (..) - No control characters - Valid path segment format

Parameters:

uri (str) – The kref URI to validate.

Raises:

KrefValidationError – If the URI is invalid or contains malicious patterns.

Return type:

None

Example:

from kumiho.kref import validate_kref, KrefValidationError

try:
    validate_kref("kref://project/space/item.kind?r=1")
except KrefValidationError as e:
    print(f"Invalid kref: {e}")

kumiho.validate_edge_type(edge_type)

Validate an edge type for security and correctness.

Edge types must: - Start with an uppercase letter - Contain only uppercase letters, digits, and underscores - Be 1-50 characters long

Parameters:

edge_type (str) – The edge type to validate.

Raises:

EdgeTypeValidationError – If the edge type is invalid.

Return type:

None

Example:

from kumiho.edge import validate_edge_type, EdgeTypeValidationError

try:
    validate_edge_type("DEPENDS_ON")  # OK
    validate_edge_type("depends_on")  # Raises error
except EdgeTypeValidationError as e:
    print(f"Invalid edge type: {e}")

kumiho.is_valid_kref(uri)

Check if a Kref URI is valid without raising exceptions.

Parameters:

uri (str) – The kref URI to validate.

Return type:

bool

Returns:

True if the URI is valid, False otherwise.

Example:

from kumiho.kref import is_valid_kref

if is_valid_kref("kref://project/space/item.kind"):
    print("Valid!")

kumiho.is_valid_edge_type(edge_type)

Check if an edge type is valid without raising exceptions.

Parameters:

edge_type (str) – The edge type to validate.

Return type:

bool

Returns:

True if the edge type is valid, False otherwise.

kumiho.connect(endpoint=None, token=None, *, enable_auto_login=True, use_discovery=None, default_metadata=None, tenant_hint=None)

Create a new Kumiho client with explicit configuration.

Use this when you need more control over the client configuration, such as connecting to a specific server or using a custom token.

Parameters:

• endpoint (Optional[str]) – The gRPC server endpoint (e.g., “localhost:50051” or “https://us-central.kumiho.cloud”).

• token (Optional[str]) – The authentication token. If not provided and enable_auto_login is True, attempts to load from cache.

• enable_auto_login (bool) – If True, automatically use cached credentials when no token is provided.

• use_discovery (Optional[bool]) – If True, use the discovery service to find the regional server. If None, auto-detect.

• default_metadata (Optional[List[Tuple[str, str]]]) – Additional gRPC metadata to include with all requests (e.g., custom headers).

• tenant_hint (Optional[str]) – Tenant slug or ID for multi-tenant routing.

Returns:

A configured client instance.

Return type:

_Client

Example

Connect to a local development server:

client = kumiho.connect(
    endpoint="localhost:50051",
    token=None  # No auth for local dev
)

Connect to production with explicit token:

client = kumiho.connect(
    endpoint="https://us-central.kumiho.cloud",
    token=os.environ["KUMIHO_TOKEN"]
)

Use with context manager for temporary switching:

client = kumiho.connect(endpoint="localhost:50051")
with kumiho.use_client(client):
    local_projects = kumiho.get_projects()

See also

auto_configure_from_discovery(): Recommended for production. configure_default_client(): Set as global default.

class kumiho.use_client

Bases: object

Context manager to temporarily set the current client instance.

This is useful for multi-tenant scenarios or when you need to use different clients for different operations within the same thread or async context.

Parameters:

client (_Client) – The client to use within the context.

Example

Using different clients for different tenants:

import kumiho

tenant_a_client = kumiho.connect(endpoint="tenant-a.kumiho.cloud:443")
tenant_b_client = kumiho.connect(endpoint="tenant-b.kumiho.cloud:443")

with kumiho.use_client(tenant_a_client):
    # All operations here use tenant_a_client
    projects_a = kumiho.get_projects()

with kumiho.use_client(tenant_b_client):
    # All operations here use tenant_b_client
    projects_b = kumiho.get_projects()

Note

Context-local clients take precedence over the global default. This works correctly with async code and concurrent requests.

__init__(client)

Initialize the context manager.

Parameters:

client (_Client) – The client to use within the context.

__enter__()

Enter the context and set the client.

__exit__(exc_type, exc_val, exc_tb)

Exit the context and restore the previous client.

kumiho.get_client()

Get the current Kumiho client instance.

Returns the context-local client if set via use_client, otherwise returns the global default client (creating one if needed).

Returns:

The active client instance.

Return type:

_Client

Raises:

RuntimeError – If no client is configured and auto-bootstrap fails.

Example

>>> client = kumiho.get_client()
>>> projects = client.get_projects()

kumiho.configure_default_client(client)

Set the global default client used by top-level helper functions.

This function allows you to manually configure the default client that will be used by functions like create_project(), get_projects(), etc.

Parameters:

client (_Client) – The client instance to set as the default.

Returns:

The same client instance (for chaining).

Return type:

_Client

Example

>>> client = kumiho.connect(endpoint="localhost:50051")
>>> kumiho.configure_default_client(client)
>>> # Now all top-level functions use this client
>>> projects = kumiho.get_projects()

kumiho.auto_configure_from_discovery(*, tenant_hint=None, force_refresh=False, interactive=False)

Configure the default client using cached credentials and discovery.

This is the recommended way to bootstrap the Kumiho client. It uses credentials cached by kumiho-auth login and calls the control-plane discovery endpoint to resolve the correct regional server.

Parameters:

• tenant_hint (Optional[str]) – Optional tenant slug or ID to use for discovery. If not provided, the user’s default tenant is used.

• force_refresh (bool) – If True, bypass the discovery cache and fetch fresh routing information from the control plane.

• interactive (bool) – If True and no cached credentials exist, prompt for interactive login. Defaults to False for script safety.

Returns:

The configured client instance, also set as the default.

Return type:

_Client

Raises:

RuntimeError – If no cached credentials exist and interactive mode is disabled.

Example

Basic usage:

import kumiho

# First, run: kumiho-auth login
# Then in your code:
kumiho.auto_configure_from_discovery()

# Now you can use all kumiho functions
projects = kumiho.get_projects()

With tenant hint for multi-tenant access:

kumiho.auto_configure_from_discovery(tenant_hint="my-studio")

See also

connect(): For manual client configuration. use_client: For temporary client switching.

kumiho.LATEST_TAG = 'latest'

Standard tag name indicating the latest revision of an item.

Type:

str

kumiho.PUBLISHED_TAG = 'published'

Standard tag name indicating a published/released revision.

Type:

str

class kumiho.EdgeType

Bases: object

Standard edge types for Kumiho relationships.

These constants define the semantic meaning of relationships between revisions. Use them when creating or querying edges.

All edge types use UPPERCASE format as required by the Neo4j graph database.

BELONGS_TO

Indicates ownership or grouping relationship.

Type:

str

CREATED_FROM

Indicates the source was generated from target.

Type:

str

REFERENCED

Indicates a soft reference relationship.

Type:

str

DEPENDS_ON

Indicates the source requires the target.

Type:

str

DERIVED_FROM

Indicates the source was derived/modified from target.

Type:

str

CONTAINS

Indicates the source contains or includes the target.

Type:

str

Example:

import kumiho

# Model depends on texture
model_v1.create_edge(texture_v2, kumiho.DEPENDS_ON)

# LOD derived from high-poly
lod_v1.create_edge(highpoly_v1, kumiho.DERIVED_FROM)

BELONGS_TO = 'BELONGS_TO'

Ownership or grouping relationship.

CONTAINS = 'CONTAINS'

Source contains or includes target.

CREATED_FROM = 'CREATED_FROM'

Source was generated/created from target.

DEPENDS_ON = 'DEPENDS_ON'

Source requires target to function.

DERIVED_FROM = 'DERIVED_FROM'

Source was derived or modified from target.

REFERENCED = 'REFERENCED'

Soft reference relationship.

class kumiho.EdgeDirection

Bases: object

Direction constants for edge traversal queries.

When querying edges, you can specify which direction to traverse: outgoing edges (from source), incoming edges (to target), or both.

OUTGOING

Edges where the queried revision is the source.

Type:

int

INCOMING

Edges where the queried revision is the target.

Type:

int

BOTH

Edges in either direction.

Type:

int

Example:

import kumiho

# Get dependencies (what this revision depends on)
deps = revision.get_edges(kumiho.DEPENDS_ON, kumiho.OUTGOING)

# Get dependents (what depends on this revision)
dependents = revision.get_edges(kumiho.DEPENDS_ON, kumiho.INCOMING)

# Get all relationships
all_edges = revision.get_edges(direction=kumiho.BOTH)

BOTH = 2

Edges in either direction.

INCOMING = 1

Edges where the queried revision is the target.

OUTGOING = 0

Edges where the queried revision is the source.

kumiho.get_tenant_info(tenant_hint=None)

Get information about the current tenant from the discovery cache.

This retrieves tenant information that was cached during discovery, including the tenant ID, tenant name/slug, roles, and region info.

Parameters:

tenant_hint (Optional[str]) – Optional tenant slug or ID to look up. If not provided, looks up the default tenant (using “_default” cache key).

Returns:

A dictionary containing tenant info:

• tenant_id: The unique tenant identifier

• tenant_name: The tenant slug/name (human-readable identifier)

• roles: List of roles the user has in this tenant

• region: Region routing information

Returns None if no cached info is available.

Return type:

Optional[Dict[str, Any]]

Example

>>> info = kumiho.get_tenant_info()
>>> if info:
...     print(f"Tenant: {info['tenant_name']}")
...     print(f"Roles: {info['roles']}")
Tenant: kumihoclouds
Roles: ['owner', 'admin']

Note

This requires that discovery has been performed (either via auto_configure_from_discovery() or KUMIHO_AUTO_CONFIGURE=1). If using direct connection without discovery, this will return None.

kumiho.get_tenant_slug(tenant_hint=None)

Get the tenant slug/name for use in project naming.

This is a convenience function that returns just the tenant name/slug, which is useful for constructing project names like “ComfyUI@{tenant_slug}”.

If tenant_name contains spaces or special characters, returns a shortened tenant_id instead (first 8 characters of the UUID).

Parameters:

tenant_hint (Optional[str]) – Optional tenant slug or ID to look up.

Returns:

The tenant slug/name, or None if not available.

Return type:

Optional[str]

Example

>>> slug = kumiho.get_tenant_slug()
>>> project_name = f"ComfyUI@{slug or 'default'}"
>>> print(project_name)
ComfyUI@kumihoclouds

kumiho.create_project(name, description='')

Create a new project.

Projects are the top-level containers for all assets. Each project has its own namespace for spaces and items.

Parameters:

• name (str) – The unique name for the project. Must be URL-safe (lowercase letters, numbers, hyphens).

• description (str) – Optional human-readable description.

Returns:

The newly created Project object.

Return type:

Project

Raises:

• ProjectLimitError – If the tenant has reached their project limit.

• KumihoError – If the project name is invalid or already exists.

Example

>>> project = kumiho.create_project(
...     "commercial-2024",
...     "Assets for 2024 commercial campaign"
... )
>>> print(project.name)
commercial-2024

kumiho.get_projects()

List all projects accessible to the current user.

Returns:

A list of Project objects.

Return type:

List[Project]

Example

>>> projects = kumiho.get_projects()
>>> for p in projects:
...     print(f"{p.name}: {p.description}")
commercial-2024: Assets for 2024 commercial campaign
film-project: Feature film VFX assets

kumiho.get_project(name)

Get a project by name.

Parameters:

name (str) – The name of the project to retrieve.

Returns:

The Project object if found, None otherwise.

Return type:

Optional[Project]

Example

>>> project = kumiho.get_project("commercial-2024")
>>> if project:
...     spaces = project.get_spaces()

kumiho.delete_project(project_id, force=False)

Delete a project.

Parameters:

• project_id (str) – The unique ID of the project to delete.

• force (bool) – If True, permanently delete the project and all its contents. If False (default), mark as deprecated.

Returns:

A StatusResponse indicating success or failure.

Return type:

StatusResponse

Warning

Force deletion is irreversible and removes all spaces, items, revisions, artifacts, and edges within the project.

Example

>>> # Soft delete (deprecate)
>>> kumiho.delete_project("proj-uuid-here")

>>> # Hard delete (permanent)
>>> kumiho.delete_project("proj-uuid-here", force=True)

kumiho.item_search(context_filter='', name_filter='', kind_filter='')

Search for items across projects and spaces.

Parameters:

• context_filter (str) – Filter by project or space path. Supports glob patterns like project-* or */characters/*.

• name_filter (str) – Filter by item name. Supports wildcards.

• kind_filter (str) – Filter by item kind (e.g., “model”, “texture”).

Returns:

A list of Item objects matching the filters.

Return type:

List[Item]

Example

>>> # Find all models in any project
>>> models = kumiho.item_search(kind_filter="model")

>>> # Find character assets in a specific project
>>> chars = kumiho.item_search(
...     context_filter="film-project/characters",
...     kind_filter="model"
... )

>>> # Wildcard search
>>> heroes = kumiho.item_search(name_filter="hero*")

kumiho.get_item(kref)

Get an item by its kref URI.

Parameters:

kref (str) – The kref URI of the item (e.g., “kref://project/space/item.kind”).

Returns:

The Item object.

Return type:

Item

Raises:

grpc.RpcError – If the item is not found.

Example

>>> item = kumiho.get_item(
...     "kref://film-project/characters/hero.model"
... )
>>> revisions = item.get_revisions()

kumiho.get_bundle(kref)

Get a bundle by its kref URI.

This is a convenience function that gets an item and verifies it’s a bundle.

Parameters:

kref (str) – The kref URI of the bundle item (e.g., “kref://project/space/bundle-name.bundle”).

Returns:

The Bundle object.

Return type:

Bundle

Raises:

• ValueError – If the item exists but is not a bundle.

• grpc.RpcError – If the bundle is not found.

Example

>>> bundle = kumiho.get_bundle(
...     "kref://film-project/shots/shot001.bundle"
... )
>>> members = bundle.get_members()

kumiho.get_revision(kref)

Get a revision by its kref URI.

Parameters:

kref (str) – The kref URI of the revision (e.g., “kref://project/space/item.kind?r=1”).

Returns:

The Revision object.

Return type:

Revision

Raises:

grpc.RpcError – If the revision is not found.

Example

>>> revision = kumiho.get_revision(
...     "kref://film-project/characters/hero.model?r=3"
... )
>>> artifacts = revision.get_artifacts()
>>> for a in artifacts:
...     print(a.location)

kumiho.get_artifact(kref)

Get an artifact by its kref URI.

Parameters:

kref (str) – The kref URI of the artifact (e.g., “kref://project/space/item.kind?r=1&a=main”).

Returns:

The Artifact object.

Return type:

Artifact

Raises:

• grpc.RpcError – If the artifact is not found.

• ValueError – If the kref is missing the artifact name (&a=).

Example

>>> artifact = kumiho.get_artifact(
...     "kref://film-project/characters/hero.model?r=3&a=mesh"
... )
>>> print(artifact.location)
/projects/film/char/hero_v3.fbx

kumiho.get_artifacts_by_location(location)

Find all artifacts at a specific file location.

This is useful for reverse lookups—finding which Kumiho artifacts reference a particular file path.

Parameters:

location (str) – The file path or URI to search for.

Returns:

A list of Artifact objects at that location.

Return type:

List[Artifact]

Example

>>> artifacts = kumiho.get_artifacts_by_location(
...     "/shared/assets/hero_v3.fbx"
... )
>>> for a in artifacts:
...     print(f"{a.kref} -> {a.location}")

kumiho.set_attribute(kref, key, value)

Set a single metadata attribute on any entity.

This allows granular updates to metadata without replacing the entire metadata map. Works on any entity type (Revision, Item, Artifact, or Space) identified by kref.

Parameters:

• kref (str) – The kref URI of the entity.

• key (str) – The attribute key to set.

• value (str) – The attribute value.

Returns:

True if the attribute was set successfully.

Return type:

bool

Raises:

grpc.RpcError – If the entity is not found or the key is reserved.

Example

>>> kumiho.set_attribute(
...     "kref://project/models/hero.model?r=1",
...     "render_engine",
...     "cycles"
... )
True

kumiho.get_attribute(kref, key)

Get a single metadata attribute from any entity.

Parameters:

• kref (str) – The kref URI of the entity.

• key (str) – The attribute key to retrieve.

Return type:

Optional[str]

Returns:

The attribute value if it exists, None otherwise.

Example

>>> kumiho.get_attribute(
...     "kref://project/models/hero.model?r=1",
...     "render_engine"
... )
"cycles"

kumiho.delete_attribute(kref, key)

Delete a single metadata attribute from any entity.

Parameters:

• kref (str) – The kref URI of the entity.

• key (str) – The attribute key to delete.

Returns:

True if the attribute was deleted successfully.

Return type:

bool

Raises:

grpc.RpcError – If the entity is not found or the key is reserved.

Example

>>> kumiho.delete_attribute(
...     "kref://project/models/hero.model?r=1",
...     "old_field"
... )
True

kumiho.event_stream(routing_key_filter='', kref_filter='', cursor=None, consumer_group=None, from_beginning=False)

Subscribe to real-time events from the Kumiho server.

Events are streamed as they occur, allowing you to react to changes in the database such as new revisions, tag changes, or deletions.

Parameters:

• routing_key_filter (str) – Filter events by routing key pattern. Supports wildcards (e.g., item.model.*, revision.#).

• kref_filter (str) – Filter events by kref pattern. Supports glob patterns (e.g., kref://projectA/**/*.model).

• cursor (Optional[str]) – Resume from a previous cursor position (Creator tier+). Pass the cursor from the last received event to continue from that point after reconnection.

• consumer_group (Optional[str]) – Consumer group name for load-balanced delivery (Enterprise tier only). Multiple consumers in the same group each receive different events.

• from_beginning (bool) – Start from earliest available events instead of live-only (Creator tier+, subject to retention limits).

Yields:

Event –

Event objects as they occur. Each event includes a cursor

field that can be saved for resumption.

Return type:

Iterator[Event]

Example

>>> # Watch for all revision events with cursor tracking
>>> last_cursor = None
>>> try:
...     for event in kumiho.event_stream(
...         routing_key_filter="revision.*",
...         kref_filter="kref://film-project/**",
...         cursor=last_cursor
...     ):
...         print(f"{event.routing_key}: {event.kref}")
...         last_cursor = event.cursor  # Save for reconnection
...         if event.routing_key == "revision.tagged":
...             print(f"  Tag: {event.details.get('tag')}")
... except ConnectionError:
...     # Reconnect using saved cursor
...     pass

Note

This is a blocking iterator. Use in a separate thread or async context for production applications.

Cursor-based resume requires Creator tier or above. Use get_event_capabilities() to check your tier’s capabilities.

kumiho.get_event_capabilities()

Get event streaming capabilities for the current tenant tier.

Returns the capabilities available based on the authenticated tenant’s subscription tier. Use this to determine which features (cursor resume, consumer groups, replay) are available before using them.

Returns:

Object with capability flags and limits:

• supports_replay: Can replay past events

• supports_cursor: Can resume from cursor

• supports_consumer_groups: Can use consumer groups (Enterprise)

• max_retention_hours: Event retention period (-1 = unlimited)

• max_buffer_size: Max events in buffer (-1 = unlimited)

• tier: Tier name (free, creator, studio, enterprise)

Return type:

EventCapabilities

Example

>>> caps = kumiho.get_event_capabilities()
>>> print(f"Tier: {caps.tier}")
Tier: creator
>>> if caps.supports_cursor:
...     # Use cursor-based streaming
...     last_cursor = load_saved_cursor()
...     for event in kumiho.event_stream(cursor=last_cursor):
...         process(event)
...         save_cursor(event.cursor)
... else:
...     # Free tier - no cursor support
...     for event in kumiho.event_stream():
...         process(event)

class kumiho.EventCapabilities

Bases: object

Event streaming capabilities for the current tenant tier.

supports_replay

Whether this tier supports replaying past events.

supports_cursor

Whether cursor-based resume is supported.

supports_consumer_groups

Whether consumer groups are supported (Enterprise only).

max_retention_hours

Maximum event retention in hours (0 = none, -1 = unlimited).

max_buffer_size

Maximum events in buffer (0 = none, -1 = unlimited).

tier

Tier identifier (free, creator, studio, enterprise).

__init__(supports_replay, supports_cursor, supports_consumer_groups, max_retention_hours, max_buffer_size, tier)

Parameters:

• supports_replay (bool)

• supports_cursor (bool)

• supports_consumer_groups (bool)

• max_retention_hours (int)

• max_buffer_size (int)

• tier (str)

Return type:

None

supports_replay: bool

supports_cursor: bool

supports_consumer_groups: bool

max_retention_hours: int

max_buffer_size: int

tier: str

kumiho.resolve(kref)

Resolve a kref URI to a file location.

This is a convenience function to get the file path for an artifact or the default artifact of a revision.

Parameters:

kref (str) – The kref URI to resolve.

Returns:

The file location string, or None if not resolvable.

Return type:

Optional[str]

Example

>>> # Resolve a specific artifact
>>> path = kumiho.resolve(
...     "kref://film-project/chars/hero.model?r=3&a=mesh"
... )
>>> print(path)
/projects/film/char/hero_v3.fbx

>>> # Resolve revision's default artifact
>>> path = kumiho.resolve(
...     "kref://film-project/chars/hero.model?r=3"
... )