in aws_advanced_python_wrapper/plugin_service.py [0:0]
def __init__(
self,
container: PluginServiceManagerContainer,
props: Properties,
target_func: Callable,
driver_dialect_manager: DriverDialectManager,
driver_dialect: DriverDialect,
session_state_service: Optional[SessionStateService] = None):
self._container = container
self._container.plugin_service = self
self._props = props
self._original_url = PropertiesUtils.get_url(props)
self._host_list_provider: HostListProvider = ConnectionStringHostListProvider(self, props)
self._all_hosts: Tuple[HostInfo, ...] = ()
self._allowed_and_blocked_hosts: Optional[AllowedAndBlockedHosts] = None
self._current_connection: Optional[Connection] = None
self._current_host_info: Optional[HostInfo] = None
self._initial_connection_host_info: Optional[HostInfo] = None
self._exception_manager: ExceptionManager = ExceptionManager()
self._is_in_transaction: bool = False
self._dialect_provider = DatabaseDialectManager(props)
self._target_func = target_func
self._driver_dialect_manager = driver_dialect_manager
self._driver_dialect = driver_dialect
self._database_dialect = self._dialect_provider.get_dialect(driver_dialect.dialect_code, props)
self._session_state_service = session_state_service if session_state_service is not None else SessionStateServiceImpl(self, props)