INNER CODE UNIT · Python

_auto_connect_clickhouse

infobyte/emploleaks · emploleaks.py:604

    def _auto_connect_clickhouse(self):
        """Try to connect to ClickHouse using saved config. Fails silently."""
        try:
            self.leaks_conn = clickhouse_driver.Client(
                host=self.ch_host,
                port=int(self.ch_port),
                user=self.ch_user,
                password=self.ch_passwd,
            )
            self.leaks_conn.execute('SELECT 1')
            self._ensure_credentials_db()
            log.debug(f"ClickHouse auto-connected to {self.ch_host}:{self.ch_port}")
        except Exception:
            self.leaks_conn = None

    def _ensure_credentials_db(self):
        """Create the credentials database and table if they don't exist."""
        if not self.leaks_conn:

View source record →