| | import threading |
| | from abc import ABC, abstractmethod |
| |
|
| | import tinydb |
| | import tinymongo as tm |
| |
|
| |
|
| | class AbstractDatabaseDocument(ABC): |
| |
|
| | @abstractmethod |
| | def _save(self, key, data): |
| | '''Save the data in the database''' |
| | pass |
| |
|
| | @abstractmethod |
| | def _get(self, key): |
| | '''Get the data from the database''' |
| | pass |
| |
|
| | @abstractmethod |
| | def _getId(self): |
| | '''Get the id of the document''' |
| | pass |
| |
|
| | @abstractmethod |
| | def __str__(self): |
| | '''Return the string representation of the document''' |
| | pass |
| |
|
| | @abstractmethod |
| | def _delete(self): |
| | '''Delete the document''' |
| | pass |
| |
|
| |
|
| | class TinyMongoClient(tm.TinyMongoClient): |
| | @property |
| | def _storage(self): |
| | return tinydb.storages.JSONStorage |
| |
|
| |
|
| | TINY_MONGO_DATABASE = TinyMongoClient("./.database") |
| |
|
| |
|
| | class TinyMongoDocument(AbstractDatabaseDocument): |
| | _lock = threading.Lock() |
| |
|
| | def __init__(self, db_name: str, collection_name: str, document_id: str, create=False): |
| | self.collection = TINY_MONGO_DATABASE[db_name][collection_name] |
| | self.collection_name = collection_name |
| | self.document_id = document_id |
| | if (not self.exists()): |
| | if create: |
| | self.collection.insert_one({"_id": document_id}) |
| | else: |
| | raise Exception(f"The document with id {document_id} in collection {collection_name} of database {db_name} does not exist") |
| |
|
| | def exists(self): |
| | with self._lock: |
| | return self.collection.find({"_id": self.document_id}).count() == 1 |
| |
|
| | def _save(self, data): |
| | with self._lock: |
| | try: |
| | update_data = {'$set': {}} |
| | for key, value in data.items(): |
| | path_parts = key.split(".") |
| |
|
| | if len(path_parts) > 1: |
| | root_key = ".".join(path_parts[:-1]) |
| | last_key = path_parts[-1] |
| | current_value = self._get(root_key) |
| | if not isinstance(current_value, dict): |
| | current_value = {} |
| | current_value[last_key] = value |
| | update_data['$set'][root_key] = current_value |
| | else: |
| | update_data['$set'][key] = value |
| |
|
| | self.collection.update_one({'_id': self.document_id}, update_data) |
| | except Exception as e: |
| | print(f"Error saving data: {e}") |
| |
|
| | def _get(self, key=None): |
| | with self._lock: |
| | try: |
| | document = self.collection.find_one({'_id': self.document_id}) |
| | if not key: |
| | del document['_id'] |
| | return document |
| | keys = key.split(".") |
| | value = document[keys[0]] |
| | for k in keys[1:]: |
| | value = value[k] |
| | return value |
| | except Exception as e: |
| | |
| | return None |
| |
|
| | def _delete(self, key): |
| | with self._lock: |
| | try: |
| | document = self.collection.find_one({'_id': self.document_id}) |
| | if key in document: |
| | del document[key] |
| | self.collection.remove({'_id': self.document_id}) |
| | self.collection.insert(document) |
| | else: |
| | print(f"Key '{key}' not found in the document") |
| | except Exception as e: |
| | print(f"Error deleting key '{key}': {e}") |
| |
|
| | def _getId(self): |
| | return self.document_id |
| |
|
| | def __str__(self): |
| | with self._lock: |
| | document = self.collection.find_one({'_id': self.document_id}) |
| | return str(document) |
| |
|
