Method connect

xtop/core/data_source.py:38–45

Get or create DuckDB connection

(self)

            raise ValueError(f"Data directory does not exist: {datadir}")
    
    def connect(self):
        """Get or create DuckDB connection"""
        if self.conn is None:
            self.conn = duckdb.connect(':memory:')
            # Configure thread count if specified
            if self.duckdb_threads is not None:
                self.conn.execute(f"SET threads TO {self.duckdb_threads}")
        return self.conn
    
    def close(self):
        """Close DuckDB connection"""