def check_connection()

in src/databao_context_engine/plugins/resources/parquet_introspector.py [0:0]

• 12 lines of code
• 7 McCabe index (conditional complexity)

    def check_connection(self, file_config: ParquetConfigFile) -> None:
        with self._connect(file_config) as conn:
            with conn.cursor() as cur:
                resolved_url = _resolve_url(file_config)
                cur.execute(f"SELECT * FROM parquet_file_metadata('{resolved_url}') LIMIT 1")
                columns = [desc[0].lower() for desc in cur.description] if cur.description else []
                rows = cur.fetchall()
                parquet_file_metadata = [dict(zip(columns, row)) for row in rows]
                if not parquet_file_metadata:
                    raise ValueError(f"No parquet files found by url {resolved_url}")
                if not parquet_file_metadata or not parquet_file_metadata[0]["file_name"]:
                    raise ValueError("Parquet resource introspection failed")