From fd899f554985cecdfd5da04e34decd33063f1bcb Mon Sep 17 00:00:00 2001 From: Steve Nyemba Date: Mon, 1 Apr 2024 12:52:42 -0500 Subject: [PATCH] adding wrapper class/design pattern to support plugins --- transport/iowrapper.py | 47 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 transport/iowrapper.py diff --git a/transport/iowrapper.py b/transport/iowrapper.py new file mode 100644 index 0000000..f113d85 --- /dev/null +++ b/transport/iowrapper.py @@ -0,0 +1,47 @@ +""" +This class is a wrapper around read/write classes of cloud,sql,nosql,other packages +The wrapper allows for application of plugins as pre-post conditions +""" +class IO: + """ + Base wrapper class for read/write + """ + def __init__(self,_agent,loader): + self._agent = _agent + self._loader = loader + def meta (self,**_args): + if hasattr(self._agent,'meta') : + return self._agent.meta(**_args) + return [] + + def close(self): + if hasattr(self._agent,'close') : + self._agent.close() + def apply(self): + """ + applying pre/post conditions given a pipeline expression + """ + for _pointer in self._loader : + _data = _pointer(_data) + def apply(self,_query): + if hasattr(self._agent,'apply') : + return self._agent.apply(_query) + return None +class IReader(IO): + def __init__(self,_agent,pipeline=None): + super().__init__(_agent,pipeline) + def read(self,**_args): + _data = self._agent.read(**_args) + if self._loader and self._loader.ratio() > 0 : + _data = self._loader.apply(_data) + # + # output data + return _data +class IWriter(IO): + def __init__(self,_agent,pipeline=None): + super().__init__(_agent,pipeline) + def write(self,_data,**_args): + if self._loader and self._loader.ratio() > 0 : + _data = self._loader.apply(_data) + + self._agent.write(_data,**_args)