Files
nimbusflow/backend/database/connection.py

25 lines
699 B
Python

import sqlite3
from typing import List, Tuple
class DatabaseConnection:
def __init__(self, db_name: str):
self.conn = sqlite3.connect(db_name)
self.cursor = self.conn.cursor()
def close_connection(self):
self.conn.close()
def execute_query(self, query: str, params: Tuple = None):
if params:
self.cursor.execute(query, params)
else:
self.cursor.execute(query)
self.conn.commit()
def execute_query_with_return(self, query: str, params: Tuple = None):
if params:
self.cursor.execute(query, params)
else:
self.cursor.execute(query)
return self.cursor.fetchall()