Spaces:
Running
on
Zero
Running
on
Zero
File size: 1,661 Bytes
b2add11 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
import duckdb
class DuckDBModel:
def __init__(self, db_file):
self.conn = duckdb.connect(database=db_file)
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, conditions=None):
query = f"SELECT {', '.join(columns)} FROM {table_name}"
if conditions:
query += f" WHERE {conditions}"
return self.conn.execute(query).fetchall()
def update(self, table_name, columns, values, conditions):
if not conditions:
raise ValueError("Conditions are required for update operations.")
query = f"UPDATE {table_name} SET {', '.join([f'{col} = ?' for col in columns])} WHERE {conditions}"
self.conn.execute(query, values)
def delete(self, table_name, conditions):
if not conditions:
raise ValueError("Conditions are required for delete operations.")
query = f"DELETE FROM {table_name} WHERE {conditions}"
self.conn.execute(query)
# 例としての使用方法
if __name__ == "__main__":
db = DuckDBModel("test.db")
db.create_table("users", ["id INTEGER", "name TEXT"])
db.insert("users", [1, "Alice"])
print(db.select("users", ["*"]))
db.update("users", ["name"], ["Alice Smith"], "id = 1")
print(db.select("users", ["*"]))
db.delete("users", "id = 1")
print(db.select("users", ["*"]))
|