Created
October 24, 2024 09:16
-
-
Save quitegreensky/66c9f865bcfe0505a078808f581f7d68 to your computer and use it in GitHub Desktop.
Simple key, val structured database using sqlite3 for basic usage
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import sqlite3 | |
| import json | |
| import threading | |
| class JSONLikeSQLite: | |
| def __init__(self, db_file='database.sqlite'): | |
| self.db_file = db_file | |
| self.lock = threading.Lock() # Lock for managing concurrent access | |
| self.initialize_db() | |
| def initialize_db(self): | |
| with self.lock: | |
| with sqlite3.connect(self.db_file) as conn: | |
| cursor = conn.cursor() | |
| cursor.execute(''' | |
| CREATE TABLE IF NOT EXISTS json_data ( | |
| key TEXT PRIMARY KEY NOT NULL, | |
| value TEXT NOT NULL | |
| ) | |
| ''') | |
| conn.commit() | |
| def set_param(self, key, data: dict): | |
| """ Set or update a nested parameter. """ | |
| # Retrieve the current dictionary or initialize a new one | |
| current_dict = self[key] # Returns an empty dict if the key doesn't exist | |
| # Update the current dictionary with the new data | |
| current_dict.update(data) | |
| # Save the updated dictionary back to the database | |
| self[key] = current_dict | |
| def get_param(self, key, field): | |
| """ Get a specific field from a nested dictionary. """ | |
| current_dict = self[key] # This will return {} if the key doesn't exist | |
| return current_dict.get(field, None) | |
| def __setitem__(self, key, value): | |
| """ Set a key-value pair (insert/update). """ | |
| with self.lock: | |
| with sqlite3.connect(self.db_file) as conn: | |
| cursor = conn.cursor() | |
| cursor.execute(''' | |
| INSERT INTO json_data (key, value) VALUES (?, ?) | |
| ON CONFLICT(key) DO UPDATE SET value = excluded.value | |
| ''', (key, json.dumps(value))) | |
| conn.commit() | |
| def __getitem__(self, key): | |
| """ Get the value for a given key. """ | |
| with self.lock: | |
| with sqlite3.connect(self.db_file) as conn: | |
| cursor = conn.cursor() | |
| cursor.execute(''' | |
| SELECT value FROM json_data WHERE key = ? | |
| ''', (key,)) | |
| result = cursor.fetchone() | |
| return json.loads(result[0]) if result else {} | |
| def __delitem__(self, key): | |
| """ Delete a key-value pair. """ | |
| with self.lock: | |
| with sqlite3.connect(self.db_file) as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('DELETE FROM json_data WHERE key = ?', (key,)) | |
| conn.commit() | |
| def keys(self): | |
| """ Return a list of keys. """ | |
| with self.lock: | |
| with sqlite3.connect(self.db_file) as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('SELECT key FROM json_data') | |
| return [row[0] for row in cursor.fetchall()] | |
| def items(self): | |
| """ Return a list of key-value pairs. """ | |
| with self.lock: | |
| with sqlite3.connect(self.db_file) as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('SELECT key, value FROM json_data') | |
| return [(key, json.loads(value)) for key, value in cursor.fetchall()] | |
| def to_dict(self): | |
| """ Return the entire database as a dictionary. """ | |
| return dict(self.items()) | |
| def delete(self, key): | |
| self.__delitem__(key) | |
| def get_item(self, key): | |
| return self.__getitem__(key) | |
| # Example usage | |
| if __name__ == '__main__': | |
| db = JSONLikeSQLite() | |
| db.set_param('user', {'name': 'Alice', 'age': 30}) | |
| print(db.get_param('user', 'age')) | |
| # Update the user's age | |
| db.set_param('user', {'age': 13}) | |
| print(db.get_param('user', 'age')) | |
| # Get a item | |
| print(db.get_item('user')) | |
| # Get all items | |
| print(db.items()) | |
| # Get all keys | |
| print(db.keys()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment