Concrete service factory implementations.
This module provides concrete implementations of service factories that
create and configure service instances according to their interfaces.
ConcreteServiceFactory(container)
Concrete implementation of service factory.
Source code in ccproxy/services/factories.py
| def __init__(self, container: ServiceContainer) -> None:
"""Initialize the service factory."""
self._container = container
|
Register all services with the container.
Source code in ccproxy/services/factories.py
| def register_services(self) -> None:
"""Register all services with the container."""
self._container.register_service(
MockResponseHandler, factory=self.create_mock_handler
)
self._container.register_service(
StreamingHandler, factory=self.create_streaming_handler
)
self._container.register_service(
ProxyConfiguration, factory=self.create_proxy_config
)
self._container.register_service(
httpx.AsyncClient, factory=self.create_http_client
)
self._container.register_service(
CLIDetectionService, factory=self.create_cli_detection_service
)
self._container.register_service(
HTTPPoolManager, factory=self.create_http_pool_manager
)
self._container.register_service(
ResponseCache, factory=self.create_response_cache
)
self._container.register_service(
BinaryResolver, factory=self.create_binary_resolver
)
self._container.register_service(
FormatRegistry, factory=self.create_format_registry
)
# Registries
self._container.register_service(
HookRegistry, factory=self.create_hook_registry
)
self._container.register_service(
oauth_registry_module.OAuthRegistry, factory=self.create_oauth_registry
)
self._container.register_service(
TaskRegistry, factory=self.create_task_registry
)
self._container.register_service(
AuthManagerRegistry, factory=self.create_auth_manager_registry
)
# Register background thread manager for hooks
self._container.register_service(
BackgroundHookThreadManager,
factory=self.create_background_hook_thread_manager,
)
from ccproxy.core.async_task_manager import AsyncTaskManager
self._container.register_service(
AsyncTaskManager,
factory=self.create_async_task_manager,
)
|
Create mock handler instance.
Source code in ccproxy/services/factories.py
| def create_mock_handler(self) -> MockResponseHandler:
"""Create mock handler instance."""
mock_generator = RealisticMockResponseGenerator()
settings = self._container.get_service(Settings)
# Create simple format adapter for anthropic->openai conversion (for mock responses)
openai_adapter = DictFormatAdapter(
response=convert_anthropic_to_openai_response,
name="mock_anthropic_to_openai",
)
# Configure streaming settings if needed
openai_thinking_xml = getattr(
getattr(settings, "llm", object()), "openai_thinking_xml", True
)
if hasattr(openai_adapter, "configure_streaming"):
openai_adapter.configure_streaming(openai_thinking_xml=openai_thinking_xml)
handler = MockResponseHandler(
mock_generator=mock_generator,
openai_adapter=openai_adapter,
error_rate=0.05,
latency_range=(0.5, 2.0),
)
return handler
|
create_streaming_handler()
Create streaming handler instance.
Requires HookManager to be registered before resolution to avoid
post-hoc patching of the handler.
Source code in ccproxy/services/factories.py
| def create_streaming_handler(self) -> StreamingHandler:
"""Create streaming handler instance.
Requires HookManager to be registered before resolution to avoid
post-hoc patching of the handler.
"""
hook_manager = self._container.get_service(HookManager)
handler = StreamingHandler(hook_manager=hook_manager)
return handler
|
Create proxy configuration instance.
Source code in ccproxy/services/factories.py
| def create_proxy_config(self) -> ProxyConfiguration:
"""Create proxy configuration instance."""
config = ProxyConfiguration()
return config
|
Create HTTP client instance.
Source code in ccproxy/services/factories.py
| def create_http_client(self) -> httpx.AsyncClient:
"""Create HTTP client instance."""
settings = self._container.get_service(Settings)
hook_manager = self._container.get_service(HookManager)
client = HTTPClientFactory.create_client(
settings=settings, hook_manager=hook_manager
)
logger.debug("http_client_created", category="lifecycle")
return client
|
create_cli_detection_service()
Create CLI detection service instance.
Source code in ccproxy/services/factories.py
| def create_cli_detection_service(self) -> CLIDetectionService:
"""Create CLI detection service instance."""
settings = self._container.get_service(Settings)
return CLIDetectionService(settings)
|
create_http_pool_manager()
Create HTTP pool manager instance.
Source code in ccproxy/services/factories.py
| def create_http_pool_manager(self) -> HTTPPoolManager:
"""Create HTTP pool manager instance."""
settings = self._container.get_service(Settings)
hook_manager = self._container.get_service(HookManager)
logger.debug(
"http_pool_manager_created",
has_hook_manager=hook_manager is not None,
hook_manager_type=type(hook_manager).__name__ if hook_manager else "None",
category="lifecycle",
)
return HTTPPoolManager(settings, hook_manager)
|
Create response cache instance.
Source code in ccproxy/services/factories.py
| def create_response_cache(self) -> ResponseCache:
"""Create response cache instance."""
return ResponseCache()
|
Create a BinaryResolver from settings.
Source code in ccproxy/services/factories.py
| def create_binary_resolver(self) -> BinaryResolver:
"""Create a BinaryResolver from settings."""
settings = self._container.get_service(Settings)
return BinaryResolver.from_settings(settings)
|
Create format adapter registry with core adapters pre-registered.
Pre-registers common format conversions to prevent plugin conflicts.
Plugins can still register their own plugin-specific adapters.
Source code in ccproxy/services/factories.py
| def create_format_registry(self) -> FormatRegistry:
"""Create format adapter registry with core adapters pre-registered.
Pre-registers common format conversions to prevent plugin conflicts.
Plugins can still register their own plugin-specific adapters.
"""
settings = self._container.get_service(Settings)
# Always use priority mode (latest behavior)
registry = FormatRegistry()
# Pre-register core format adapters
self._register_core_format_adapters(registry, settings)
registry.flush_all_logs()
logger.debug(
"format_registry_created",
category="format",
)
return registry
|
Create a HookRegistry instance.
Source code in ccproxy/services/factories.py
| def create_hook_registry(self) -> HookRegistry:
"""Create a HookRegistry instance."""
return HookRegistry()
|
Create an OAuthRegistry instance (imported lazily to avoid cycles).
Source code in ccproxy/services/factories.py
| def create_oauth_registry(self) -> Any:
"""Create an OAuthRegistry instance (imported lazily to avoid cycles)."""
from ccproxy.auth.oauth.registry import OAuthRegistry
return OAuthRegistry()
|
Create a TaskRegistry instance.
Source code in ccproxy/services/factories.py
| def create_task_registry(self) -> TaskRegistry:
"""Create a TaskRegistry instance."""
return TaskRegistry()
|
create_auth_manager_registry()
Create an AuthManagerRegistry instance.
Note: Auth managers are registered by their respective plugins during initialization.
Source code in ccproxy/services/factories.py
| def create_auth_manager_registry(self) -> AuthManagerRegistry:
"""Create an AuthManagerRegistry instance.
Note: Auth managers are registered by their respective plugins during initialization.
"""
return AuthManagerRegistry()
|
create_background_hook_thread_manager()
Create background hook thread manager instance.
Source code in ccproxy/services/factories.py
| def create_background_hook_thread_manager(self) -> BackgroundHookThreadManager:
"""Create background hook thread manager instance."""
manager = BackgroundHookThreadManager()
logger.debug("background_hook_thread_manager_created", category="lifecycle")
return manager
|
create_async_task_manager()
Create async task manager instance.
Source code in ccproxy/services/factories.py
| def create_async_task_manager(self) -> AsyncTaskManager:
"""Create async task manager instance."""
from ccproxy.core.async_task_manager import AsyncTaskManager
return AsyncTaskManager()
|