A lightweight version of Milvus wrapped with Python for vector similarity search in AI applications
—
Direct server lifecycle management for advanced use cases requiring custom control over the milvus-lite server process. This provides low-level access to server initialization, configuration, and process management.
Core server management class for controlling individual milvus-lite server instances.
class Server:
"""
The milvus-lite server implementation.
Manages a single milvus-lite server process with database file storage
and network configuration options.
"""
def __init__(self, db_file: str, address: Optional[str] = None):
"""
Initialize server with database file and optional network address.
Parameters:
- db_file (str): Path to local database file for data storage
- address (Optional[str]): Network address (host:port) or None for UDS
Raises:
- RuntimeError: If db_file name is invalid or too long
"""
def init(self) -> bool:
"""
Initialize server by validating paths and dependencies.
Returns:
- bool: True if initialization successful, False otherwise
"""
def start(self) -> bool:
"""
Start the milvus-lite server process.
Creates file lock, configures environment, and spawns server process.
Uses Unix Domain Socket for local communication by default.
Returns:
- bool: True if server started successfully, False otherwise
Raises:
- BlockingIOError: If database file is locked by another process
"""
def stop(self) -> None:
"""
Stop the server process and cleanup resources.
Releases file locks, terminates server process, and removes
temporary socket and lock files.
"""
def __del__(self) -> None:
"""Destructor ensures server is stopped when object is destroyed."""Usage Example:
from milvus_lite.server import Server
# Create server instance
server = Server("./my_database.db")
try:
# Initialize and start server
if not server.init():
raise RuntimeError("Server initialization failed")
if not server.start():
raise RuntimeError("Server startup failed")
# Server is now running and ready for connections
print(f"Server started with UDS path: {server.uds_path}")
# Use server (connect with pymilvus client using UDS path)
# ... your application logic ...
finally:
# Always cleanup server resources
server.stop()Read-only properties providing server configuration and status information.
@property
def milvus_bin(self) -> str:
"""Path to the milvus binary executable."""
@property
def log_level(self) -> str:
"""Log level from LOG_LEVEL environment variable (default: 'ERROR')."""
@property
def uds_path(self) -> str:
"""Unix Domain Socket path with 'unix:' prefix for client connections."""
@property
def args(self) -> List[str]:
"""Command line arguments for starting the milvus server process."""Usage Example:
server = Server("./test.db")
# Check server configuration before starting
print(f"Binary path: {server.milvus_bin}")
print(f"Log level: {server.log_level}")
print(f"UDS path: {server.uds_path}")
print(f"Server args: {server.args}")
# Server properties are available before calling start()Configure the server for TCP network access instead of Unix Domain Sockets.
# TCP network server
server = Server(
db_file="./network_db.db",
address="localhost:19530" # TCP address instead of UDS
)Usage Example:
# Create server with network access
server = Server("./shared_db.db", address="0.0.0.0:19530")
if server.init() and server.start():
print("Server available at localhost:19530")
# Connect with pymilvus using network address
from pymilvus import MilvusClient
client = MilvusClient(uri="http://localhost:19530")
# Use client normally...
client.create_collection("test", dimension=128)Server operations can raise various exceptions depending on the failure mode.
# Exceptions that may be raised:
# - RuntimeError: Invalid database names, missing binaries, startup failures
# - BlockingIOError: Database file locked by another process
# - subprocess.TimeoutExpired: Process management timeouts
# - FileNotFoundError: Missing required files or directoriesUsage Example:
from milvus_lite.server import Server
import logging
def start_server_safely(db_path: str) -> Optional[Server]:
"""Start server with comprehensive error handling."""
server = Server(db_path)
try:
if not server.init():
logging.error("Server initialization failed - check binary paths")
return None
if not server.start():
logging.error("Server startup failed - check logs and ports")
return None
logging.info(f"Server started successfully at {server.uds_path}")
return server
except BlockingIOError:
logging.error(f"Database {db_path} is already in use by another process")
return None
except RuntimeError as e:
logging.error(f"Runtime error starting server: {e}")
return None
except Exception as e:
logging.error(f"Unexpected error: {e}")
return NoneDatabase files must meet specific naming and path requirements for security and compatibility.
# Valid database file names (examples):
# - "data.db"
# - "vectors_2024.db"
# - "user-embeddings.db"
# - "test_collection.db"
# Invalid database file names:
# - "data base.db" (spaces not allowed)
# - "data@base.db" (special characters not allowed)
# - "very_long_database_name_that_exceeds_limits.db" (too long)File Name Rules:
^[a-zA-Z0-9.\-_]+$The Server class manages the underlying milvus binary process with proper environment setup and resource cleanup.
Environment Variables:
BIN_PATH: Override default binary path locationLOG_LEVEL: Set server log level (ERROR, WARN, INFO, DEBUG)LD_LIBRARY_PATH: Managed automatically for LinuxDYLD_LIBRARY_PATH: Managed automatically for macOSProcess Lifecycle:
Resource Cleanup:
__del__)stop() methodInstall with Tessl CLI
npx tessl i tessl/pypi-milvus-lite