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
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.
Optional error message providing details about the failure.
Methods
Returns True if the status is OK (successful), False otherwise.status = Status()
if status.ok():
print("Operation succeeded")
Returns the status code.status = Status(StatusCode.INVALID_ARGUMENT, "Invalid dimension")
print(status.code()) # StatusCode.INVALID_ARGUMENT
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:
Create a success status.status = Status.OK()
assert status.ok()
Create a NOT_FOUND error status.status = Status.NotFound("Document not found")
Status.AlreadyExists(message)
Create an ALREADY_EXISTS error status.status = Status.AlreadyExists("Collection already exists")
Status.InvalidArgument(message)
Create an INVALID_ARGUMENT error status.status = Status.InvalidArgument("Invalid vector dimension")
Status.PermissionDenied(message)
Create a PERMISSION_DENIED error status.status = Status.PermissionDenied("Write access denied")
Status.InternalError(message)
Create an INTERNAL_ERROR status.status = Status.InternalError("Database connection failed")
StatusCode Enum
Enumeration of possible status codes:
Operation completed successfully.Value: 0status = Status(StatusCode.OK)
assert status.ok()
Requested resource was not found.Value: 1Examples: Collection doesn’t exist, document not found, field not in schemaif status.code() == StatusCode.NOT_FOUND:
print("Resource not found")
Resource already exists.Value: 2Examples: Collection name conflict, duplicate primary keyif status.code() == StatusCode.ALREADY_EXISTS:
print("Resource already exists")
Invalid argument provided.Value: 3Examples: Wrong dimension, invalid data type, malformed queryif status.code() == StatusCode.INVALID_ARGUMENT:
print(f"Invalid argument: {status.message()}")
Insufficient permissions for the operation.Value: 4Examples: Read-only access, authentication failureif status.code() == StatusCode.PERMISSION_DENIED:
print("Access denied")
Operation failed because a precondition was not met.Value: 5Examples: Collection not loaded, index not builtif status.code() == StatusCode.FAILED_PRECONDITION:
print("Precondition not met")
Resource exhausted (out of memory, disk full, etc.).Value: 6Examples: Out of memory, disk quota exceededif status.code() == StatusCode.RESOURCE_EXHAUSTED:
print("Resource exhausted")
Service is unavailable.Value: 7Examples: Connection timeout, service downif status.code() == StatusCode.UNAVAILABLE:
print("Service unavailable")
Internal error occurred.Value: 8Examples: Unexpected exception, database corruptionif status.code() == StatusCode.INTERNAL_ERROR:
print(f"Internal error: {status.message()}")
Operation not supported.Value: 9Examples: Unsupported feature, deprecated APIif status.code() == StatusCode.NOT_SUPPORTED:
print("Operation not supported")
Unknown error.Value: 10Examples: Unhandled exception, unexpected stateif 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:
- Check
ok() first before accessing results
- Use factory methods (
Status.NotFound(), etc.) for clarity
- Provide descriptive messages to help debugging
- Match status codes to specific error types
- Retry only transient errors (UNAVAILABLE, RESOURCE_EXHAUSTED)
- 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
Success
Client Errors (4xx-like)
Server Errors (5xx-like)
- NOT_FOUND: Resource doesn’t exist
- ALREADY_EXISTS: Resource conflict
- INVALID_ARGUMENT: Bad input
- FAILED_PRECONDITION: Precondition not met
- PERMISSION_DENIED: Access denied
- UNAVAILABLE: Service unavailable
- RESOURCE_EXHAUSTED: Out of resources
- INTERNAL_ERROR: Internal failure
- NOT_SUPPORTED: Feature not available
- UNKNOWN: Unexpected error
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