Venice AI SDK Factory - Composition Root for Dependency Injection

This module implements the Composition Root pattern, providing a centralized factory for creating fully configured Venice AI clients with proper dependency injection. It serves as the single point where all SDK components are wired together.

The factory pattern enables:

• Centralized Configuration: Single place to configure all components
• Dependency Injection: Proper separation of concerns and testability
• Environment Management: Easy switching between production and test configurations
• Resource Management: Coordinated lifecycle management of clients

Architecture: The factory follows a specific dependency order to avoid circular references:

1. VeniceClient - Main client orchestrator
2. RateLimiter - Created after client, then injected

Key Components:

• VeniceClientFactory: Main factory class for production clients
• create_venice_client(): Convenience function for standard clients
• create_test_venice_client(): Optimized factory for testing scenarios

Example:

>>> from venice_ai.factory import VeniceClientFactory
>>> from venice_ai.core.config import VeniceAIConfig
>>>
>>> # Create production client with custom config
>>> config = VeniceAIConfig.create_production_config()
>>> client = VeniceClientFactory.create_client(
...     config=config,
...     api_key="your-api-key"
... )
>>>
>>> # Or use convenience function for simple cases
>>> from venice_ai import create_venice_client
>>> client = create_venice_client(api_key="your-api-key")

AdaptiveSchedulerAdapter Objects

class AdaptiveSchedulerAdapter()

Adapter that wraps AdaptiveScheduler to implement RateLimiterProtocol.

The AdaptiveScheduler from adaptive-rate-limiter package has a different submit_request signature (no error_factory parameter). This adapter bridges the gap by accepting error_factory but not passing it to the underlying scheduler - the AdaptiveScheduler handles error creation internally.

This follows the Adapter pattern to allow AdaptiveScheduler to be used where RateLimiterProtocol is expected.

AdaptiveSchedulerAdapter.__init__

def __init__(scheduler: Any) -> None

Initialize the adapter.

Arguments:

• scheduler - The AdaptiveScheduler instance to wrap

AdaptiveSchedulerAdapter.submit_request

async def submit_request(
        metadata: RequestMetadata,
        request_func: Callable[[], Awaitable[Any]],
        error_factory: Callable[..., Exception] | None = None) -> Any

Submit request for rate-limited execution.

Delegates to the wrapped scheduler's submit_request, ignoring error_factory since AdaptiveScheduler handles error creation internally through its mode strategies.

Arguments:

• metadata - Request metadata containing model_id, endpoint, etc.
• request_func - Async callable that executes the actual HTTP request.
• error_factory - Ignored - AdaptiveScheduler handles errors internally.

Returns:

The result from the scheduler's submit_request

AdaptiveSchedulerAdapter.is_running

def is_running() -> bool

Check if the scheduler is running.

AdaptiveSchedulerAdapter.start

async def start() -> None

Start the scheduler.

AdaptiveSchedulerAdapter.stop

async def stop() -> None

Stop the scheduler.

AdaptiveSchedulerAdapter.classifier

@property
def classifier() -> Any | None

Optional request classifier for VeniceClient compatibility.

AdaptiveSchedulerAdapter.circuit_breaker

@property
def circuit_breaker() -> Any | None

Expose circuit_breaker from underlying scheduler for health checks.

VeniceClientFactory Objects

class VeniceClientFactory()

Factory for creating fully configured Venice AI clients with dependency injection.

This class serves as the Composition Root in the dependency injection pattern, providing the single place where all Venice AI SDK dependencies are instantiated and wired together. It ensures proper configuration, initialization order, and resource management across all components.

The factory is designed to handle various deployment scenarios:

• Production environments with Redis backends
• Testing environments with optimized configurations
• Development setups with minimal dependencies
• Custom configurations for specific use cases

Design Principles:

• Single Responsibility: Only handles dependency wiring
• Configuration-Driven: All behavior controlled via VeniceAIConfig
• Environment Agnostic: Works across different deployment contexts
• Resource Safe: Proper lifecycle management and cleanup

Class Methods:

• create_client(): Main factory method for production clients
• create_test_client(): Optimized factory for testing scenarios
• create_minimal_client(): Simplified factory for basic use cases

VeniceClientFactory.create_client

@classmethod
def create_client(
        cls,
        config: VeniceAIConfig,
        api_key: str | None = None,
        account_id: str = "default",
        account_key: str | None = None,
        http_client: aiohttp.ClientSession | None = None) -> VeniceClient

Create a fully configured VeniceClient with proper dependency injection.

This is the main factory method that creates a production-ready Venice AI client with all dependencies properly wired together. The method follows a specific initialization order to avoid circular dependencies and ensure all components are properly configured.

Dependency Initialization Order:

1. VeniceClient - Main client orchestrator
2. RateLimiter - Created with client reference, then injected

Arguments:

• config - Complete Venice AI configuration containing settings for all components including backend, scheduler, rate limiting, and more.
• api_key - API key for Venice AI services. If not provided, must be set via environment variables or account_key.
• account_id - Unique identifier for this account instance. Used for multi-tenant scenarios and resource isolation.
• account_key - Account-specific API key that overrides the global api_key for this account. Defaults to api_key if not provided.
• http_client - Pre-configured aiohttp.ClientSession for HTTP requests. If not provided, the client will create its own session.

Returns:

Fully initialized VeniceClient with all dependencies injected and configured according to the provided configuration.

Raises:

• ConfigurationError - If required configuration is missing or invalid

Example:

>>> from venice_ai.factory import VeniceClientFactory
>>> from venice_ai.core.config import VeniceAIConfig
>>>
>>> config = VeniceAIConfig.create_production_config()
>>> client = VeniceClientFactory.create_client(
...     config=config,
...     api_key="your-api-key",
...     account_id="prod-account"
... )
>>>
>>> # Client is ready for use
>>> response = await client.chat.completions.create(...)

Notes:

The RateLimiter component is created with a client reference after the client is instantiated, then injected back into the client. This breaks the circular dependency between client and rate limiter.

VeniceClientFactory.create_test_client

@classmethod
def create_test_client(cls,
                       enable_redis: bool = True,
                       test_rate_multiplier: float = 10.0,
                       scheduler_mode: Any | None = None,
                       **kwargs: Any) -> VeniceClient

Create a Venice client optimized for testing.

A convenience factory method for constructing a client preconfigured for testing.

Arguments:

• enable_redis - Whether to use Redis or memory backend
• test_rate_multiplier - Rate limit multiplier for faster tests
• scheduler_mode - Optional scheduler mode (SchedulerMode enum) to use for testing. If provided, this will be passed to create_test_config().
• **kwargs - Additional arguments passed to create_client

Returns:

VeniceClient configured for testing

VeniceClientFactory.create_developer_client

@classmethod
def create_developer_client(cls,
                            *,
                            api_key: str | None = None,
                            timeout: float = 30.0,
                            max_retries: int = 1,
                            **kwargs: Any) -> VeniceClient

Create a Venice client tuned for interactive local development.

Pairs create_development_config() (memory backend, BASIC scheduler, debug logging) with fail-loud timeout/retry overrides so failures surface fast rather than waiting through production-grade backoff.

Distinct from create_test_venice_client (unit-test isolation, fake API key, faster rate multipliers) — this one expects a real API key and hits the real API.

Arguments:

• api_key - API key. If None, read from VENICE_API_KEY env var.
• timeout - HTTP request timeout in seconds. Tighter than create_development_config (60 s) so dev iteration fails loud rather than hanging.
• max_retries - HTTP retry count. 1 (one retry) makes flakes visible without removing all resilience.
• **kwargs - Forwarded to :meth:create_client.

VeniceClientFactory.create_minimal_client

@classmethod
def create_minimal_client(cls, api_key: str, **kwargs: Any) -> VeniceClient

Create a minimal Venice client for simple use cases.

Arguments:

• api_key - API key for Venice AI services
• **kwargs - Additional arguments passed to create_client

Returns:

VeniceClient with minimal configuration

create_venice_client

def create_venice_client(api_key: str,
                         config: VeniceAIConfig | None = None,
                         **kwargs: Any) -> VeniceClient

Convenience function to create a Venice client with default configuration.

Arguments:

• api_key - API key for Venice AI services
• config - Optional configuration (uses minimal config if not provided)
• **kwargs - Additional arguments passed to factory

Returns:

Configured VeniceClient instance

create_test_venice_client

def create_test_venice_client(**kwargs: Any) -> VeniceClient

Convenience function to create a Venice client for testing.

Arguments:

• **kwargs - Arguments passed to create_test_client

Returns:

VeniceClient configured for testing

create_developer_client

def create_developer_client(**kwargs: Any) -> VeniceClient

Create a Venice client tuned for interactive local development.

Convenience for :meth:VeniceClientFactory.create_developer_client — pairs create_development_config() (memory backend, BASIC scheduler, debug logging) with fail-loud timeout/retry defaults (30 s timeout, 1 retry).

Distinct from :func:create_test_venice_client (unit-test isolation): this one hits the real API with a real key. Reads VENICE_API_KEY from env when api_key is not supplied.

>>> from venice_ai import create_developer_client
>>> client = create_developer_client()  # reads VENICE_API_KEY

Arguments:

• **kwargs - Arguments forwarded to create_developer_client (api_key, timeout, max_retries, etc.).

Returns:

Configured VeniceClient.