| import duckdb | |
| class DuckDBManager: | |
| def __init__(self, db_path): | |
| self.db_path = db_path | |
| self.conn = duckdb.connect(database=self.db_path) | |
| def create_table(self, table_name, columns): | |
| query = f"CREATE TABLE IF NOT EXISTS {table_name} ({', '.join(columns)})" | |
| self.conn.execute(query) | |
| def insert(self, table_name, values): | |
| query = f"INSERT INTO {table_name} VALUES ({', '.join(['?' for _ in values])})" | |
| self.conn.execute(query, values) | |
| def select(self, table_name, columns): | |
| query = f"SELECT {', '.join(columns)} FROM {table_name}" | |
| return self.conn.execute(query).fetchall() | |
| def update(self, table_name, set_columns, where_clause): | |
| query = f"UPDATE {table_name} SET {', '.join([f'{col} = ?' for col in set_columns])} WHERE {where_clause}" | |
| self.conn.execute(query, [set_columns[col] for col in set_columns]) | |
| def delete(self, table_name, where_clause): | |
| query = f"DELETE FROM {table_name} WHERE {where_clause}" | |
| self.conn.execute(query) | |
| def close(self): | |
| self.conn.close() | |