Skip to main content

Overview

The Status class represents the outcome of Zvec operations, providing error codes and messages for debugging and error handling. It works together with StatusCode enum to indicate success or various failure conditions.
from zvec.typing import Status, StatusCode

s = Status()
print(s.ok())  # True

s = Status(StatusCode.NOT_FOUND, "Collection not found")
print(s.code() == StatusCode.NOT_FOUND)  # True
print(s.message())  # "Collection not found"

Status Class

Constructor

Status()
constructor
Create a status object representing success or failure.Overloads:
# Success status
status = Status()

# Error status
status = Status(code: StatusCode, message: str = "")

Parameters

code
StatusCode
default:"StatusCode.OK"
The status code indicating success or type of failure.
message
str
default:"\"\""
Optional error message providing details about the failure.

Methods

ok()
bool
Returns True if the status is OK (successful), False otherwise.
status = Status()
if status.ok():
    print("Operation succeeded")
code()
StatusCode
Returns the status code.
status = Status(StatusCode.INVALID_ARGUMENT, "Invalid dimension")
print(status.code())  # StatusCode.INVALID_ARGUMENT
message()
str
Returns the error message (may be empty).
status = Status(StatusCode.NOT_FOUND, "Collection 'docs' not found")
print(status.message())  # "Collection 'docs' not found"

Static Factory Methods

Convenience methods for creating common status objects:
Status.OK()
Status
Create a success status.
status = Status.OK()
assert status.ok()
Status.NotFound(message)
Status
Create a NOT_FOUND error status.
status = Status.NotFound("Document not found")
Status.AlreadyExists(message)
Status
Create an ALREADY_EXISTS error status.
status = Status.AlreadyExists("Collection already exists")
Status.InvalidArgument(message)
Status
Create an INVALID_ARGUMENT error status.
status = Status.InvalidArgument("Invalid vector dimension")
Status.PermissionDenied(message)
Status
Create a PERMISSION_DENIED error status.
status = Status.PermissionDenied("Write access denied")
Status.InternalError(message)
Status
Create an INTERNAL_ERROR status.
status = Status.InternalError("Database connection failed")

StatusCode Enum

Enumeration of possible status codes:
OK
StatusCode
Operation completed successfully.Value: 0
status = Status(StatusCode.OK)
assert status.ok()
NOT_FOUND
StatusCode
Requested resource was not found.Value: 1Examples: Collection doesn’t exist, document not found, field not in schema
if status.code() == StatusCode.NOT_FOUND:
    print("Resource not found")
ALREADY_EXISTS
StatusCode
Resource already exists.Value: 2Examples: Collection name conflict, duplicate primary key
if status.code() == StatusCode.ALREADY_EXISTS:
    print("Resource already exists")
INVALID_ARGUMENT
StatusCode
Invalid argument provided.Value: 3Examples: Wrong dimension, invalid data type, malformed query
if status.code() == StatusCode.INVALID_ARGUMENT:
    print(f"Invalid argument: {status.message()}")
PERMISSION_DENIED
StatusCode
Insufficient permissions for the operation.Value: 4Examples: Read-only access, authentication failure
if status.code() == StatusCode.PERMISSION_DENIED:
    print("Access denied")
FAILED_PRECONDITION
StatusCode
Operation failed because a precondition was not met.Value: 5Examples: Collection not loaded, index not built
if status.code() == StatusCode.FAILED_PRECONDITION:
    print("Precondition not met")
RESOURCE_EXHAUSTED
StatusCode
Resource exhausted (out of memory, disk full, etc.).Value: 6Examples: Out of memory, disk quota exceeded
if status.code() == StatusCode.RESOURCE_EXHAUSTED:
    print("Resource exhausted")
UNAVAILABLE
StatusCode
Service is unavailable.Value: 7Examples: Connection timeout, service down
if status.code() == StatusCode.UNAVAILABLE:
    print("Service unavailable")
INTERNAL_ERROR
StatusCode
Internal error occurred.Value: 8Examples: Unexpected exception, database corruption
if status.code() == StatusCode.INTERNAL_ERROR:
    print(f"Internal error: {status.message()}")
NOT_SUPPORTED
StatusCode
Operation not supported.Value: 9Examples: Unsupported feature, deprecated API
if status.code() == StatusCode.NOT_SUPPORTED:
    print("Operation not supported")
UNKNOWN
StatusCode
Unknown error.Value: 10Examples: Unhandled exception, unexpected state
if status.code() == StatusCode.UNKNOWN:
    print(f"Unknown error: {status.message()}")

Usage Examples

Basic Error Handling

from zvec import Collection
from zvec.typing import Status, StatusCode

def load_collection(name: str) -> tuple[Collection | None, Status]:
    try:
        collection = Collection(name)
        return collection, Status.OK()
    except FileNotFoundError:
        return None, Status.NotFound(f"Collection '{name}' not found")
    except PermissionError:
        return None, Status.PermissionDenied(f"No access to collection '{name}'")
    except Exception as e:
        return None, Status.InternalError(str(e))

collection, status = load_collection("documents")
if status.ok():
    print(f"Loaded collection: {collection.name}")
else:
    print(f"Error: {status.message()}")

Checking Status Codes

from zvec import Collection, Field, DataType
from zvec.typing import Status, StatusCode

def create_collection_safe(name: str, schema: list[Field]) -> Status:
    try:
        Collection.create(name=name, schema=schema)
        return Status.OK()
    except ValueError as e:
        return Status.InvalidArgument(str(e))
    except FileExistsError:
        return Status.AlreadyExists(f"Collection '{name}' already exists")
    except Exception as e:
        return Status.InternalError(str(e))

schema = [
    Field(name="id", dtype=DataType.STRING, is_primary=True),
    Field(name="vec", dtype=DataType.VECTOR_FP32, dim=768)
]

status = create_collection_safe("docs", schema)

if status.ok():
    print("Collection created successfully")
elif status.code() == StatusCode.ALREADY_EXISTS:
    print("Collection already exists, using existing one")
elif status.code() == StatusCode.INVALID_ARGUMENT:
    print(f"Invalid schema: {status.message()}")
else:
    print(f"Failed to create collection: {status.message()}")

Status Comparison

from zvec.typing import Status, StatusCode

status1 = Status.OK()
status2 = Status()
status3 = Status.NotFound("Not found")

print(status1 == status2)  # True (both OK)
print(status1 == status3)  # False (different codes)

print(status1.code() == StatusCode.OK)  # True
print(status3.code() == StatusCode.NOT_FOUND)  # True

Handling Multiple Operations

from zvec import Collection
from zvec.typing import Status, StatusCode

def batch_insert_with_status(collection: Collection, docs: list[dict]) -> Status:
    """Insert multiple documents and return aggregated status."""
    errors = []
    
    for i, doc in enumerate(docs):
        try:
            collection.insert(doc)
        except ValueError as e:
            errors.append(f"Doc {i}: {e}")
        except Exception as e:
            return Status.InternalError(f"Unexpected error at doc {i}: {e}")
    
    if errors:
        return Status.InvalidArgument(
            f"{len(errors)} documents failed: " + "; ".join(errors[:3])
        )
    
    return Status.OK()

collection = Collection("products")
documents = [
    {"id": "1", "name": "Product A", "vec": [0.1] * 768},
    {"id": "2", "name": "Product B", "vec": [0.2] * 768},
]

status = batch_insert_with_status(collection, documents)
if not status.ok():
    print(f"Batch insert failed: {status.message()}")

Retry Logic Based on Status

import time
from zvec import Collection
from zvec.typing import Status, StatusCode

def query_with_retry(
    collection: Collection,
    query_vec: list[float],
    max_retries: int = 3
) -> tuple[list, Status]:
    """Query with retry logic for transient errors."""
    
    for attempt in range(max_retries):
        try:
            results = collection.query(
                vectors={"embedding": query_vec},
                topn=10
            )
            return results, Status.OK()
        
        except ConnectionError as e:
            status = Status.Unavailable(str(e))
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            return [], status
        
        except ValueError as e:
            # Don't retry for invalid arguments
            return [], Status.InvalidArgument(str(e))
        
        except Exception as e:
            return [], Status.InternalError(str(e))
    
    return [], Status.Unavailable("Max retries exceeded")

collection = Collection("documents")
query_vector = [0.5] * 768

results, status = query_with_retry(collection, query_vector)

if status.ok():
    print(f"Found {len(results)} results")
elif status.code() == StatusCode.UNAVAILABLE:
    print(f"Service temporarily unavailable: {status.message()}")
else:
    print(f"Query failed: {status.message()}")

Best Practices

Error Handling Guidelines:
  1. Check ok() first before accessing results
  2. Use factory methods (Status.NotFound(), etc.) for clarity
  3. Provide descriptive messages to help debugging
  4. Match status codes to specific error types
  5. Retry only transient errors (UNAVAILABLE, RESOURCE_EXHAUSTED)
  6. Log error details for monitoring and debugging
Common Mistakes:
  • Ignoring status checks and assuming success
  • Using wrong status codes for error types
  • Retrying non-transient errors indefinitely
  • Not logging error messages
  • Comparing status objects instead of codes

Status Code Categories

  • OK: Operation succeeded

Retryable vs Non-Retryable

Retryable Errors

These errors may be transient and worth retrying:
  • UNAVAILABLE: Service temporarily down
  • RESOURCE_EXHAUSTED: Temporary resource constraint

Non-Retryable Errors

These errors indicate persistent problems that won’t be fixed by retrying:
  • NOT_FOUND: Resource doesn’t exist
  • ALREADY_EXISTS: Resource conflict
  • INVALID_ARGUMENT: Bad input data
  • PERMISSION_DENIED: Access control issue
  • FAILED_PRECONDITION: Requirement not met
  • NOT_SUPPORTED: Feature unavailable
  • INTERNAL_ERROR: Internal bug (may need manual intervention)

See Also

Build docs developers (and LLMs) love