Nikola Shahpazov
03/15/2023, 2:43 PMdatajoely
03/15/2023, 2:44 PMZoran
03/15/2023, 3:33 PMclass MongoDBDataSet(AbstractDataSet[Collection, Dict]):
def __init__(
self,
database: str,
collection: str,
params: Dict[str, Any] = None,
load_from_func: Dict[str, Any] = None,
save_from_func: Dict[str, Any] = None
) -> None:
self.database = database
self.collection = collection
self.params = params
self.load_from_func = load_from_func,
self.save_from_func = save_from_func,
super().__init__()
def _load(self):
mongodb_client = MongoClient(**self.params)
if self.load_from_func:
return self.call_function_from_obj(mongodb_client.get_database(self.database).get_collection(self.collection), self.load_from_func[0])
else:
return mongodb_client.get_database(self.database).get_collection(self.collection).find()
def _save(self, data: Dict) -> None:
mongodb_client = MongoClient(**self.params)
if self.save_from_func:
self.call_function_from_obj(mongodb_client.get_database(self.database).get_collection(self.collection), self.save_from_func[0])
else:
mongodb_client.get_database(self.database).get_collection(self.collection).insert_one(data)
def _exists(self) -> bool:
mongodb_client = MongoClient(**self.params)
if mongodb_client.get_database(self.database).get_collection(self.collection).find():
return True
else:
return False
def _describe(self):
return dict(database=self.database,
collection=self.collection,
params=self.params)
@staticmethod
def call_function_from_obj(obj, dict_map: Dict[Any, Any]):
def _getattr(obj, attr):
if re.match(r"(.*?)", attr):
"""Eval if only there is function call"""
return eval(f'obj.{attr}')
else:
"""Recurses through an attribute chain to get the ultimate value."""
return reduce(getattr, attr.split('.'), obj)
if dict_map and dict_map.get('func', ''):
try:
# _call = getattr(obj, f"{dict_map.get('func', '')}")
_call = _getattr(obj, f"{dict_map.get('func', '')}")
_args = dict_map.get('args', [])
_kwargs = dict_map.get('kwargs', {})
return _call(*_args, **_kwargs)
except Exception as err:
raise DataSetError(err)
else:
return None
load_from_func
save_from_func
load_from_func:
func: ''
args: []
kwargs: {}
forgot in last picture to add funcdatajoely
03/15/2023, 4:07 PMZoran
03/15/2023, 4:07 PM