hiplab/data-transport
hiplab
/
data-transport
mirror of https://github.com/lnyemba/data-transport
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

bug fix: schema (postgresql) construct

This commit is contained in:
Steve Nyemba 2025-02-24 09:26:15 -06:00
parent a1b5f2743c
commit eaa2b99a2d
2 changed files with 15 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
transport/iowrapper.py
View File

@ -45,12 +45,12 @@ class IO:
def close(self): def close(self):
if hasattr(self._agent,'close') : if hasattr(self._agent,'close') :
self._agent.close() self._agent.close()
def apply(self): # def apply(self):
""" # """
applying pre/post conditions given a pipeline expression # applying pre/post conditions given a pipeline expression
""" # """
for _pointer in self._plugins : # for _pointer in self._plugins :
_data = _pointer(_data) # _data = _pointer(_data)
def apply(self,_query): def apply(self,_query):
if hasattr(self._agent,'apply') : if hasattr(self._agent,'apply') :
return self._agent.apply(_query) return self._agent.apply(_query)

11
transport/sql/common.py
View File

@ -71,7 +71,7 @@ class Base:
@TODO: Execution of stored procedures @TODO: Execution of stored procedures
""" """
if sql.strip().lower().startswith('select') or sql.strip().lower().startswith('with') or sql.strip().startswith('show'): if sql.strip().lower().startswith('select') or sql.strip().lower().startswith('with') or sql.strip().startswith('show'):
print (self._engine)
return pd.read_sql(sql,self._engine) return pd.read_sql(sql,self._engine)
else: else:
_handler = self._engine.connect() _handler = self._engine.connect()
@ -83,6 +83,7 @@ class Base:
class SQLBase(Base): class SQLBase(Base):
def __init__(self,**_args): def __init__(self,**_args):
super().__init__(**_args) super().__init__(**_args)
self._schema = _args.get('schema',None)
def get_provider(self): def get_provider(self):
raise Exception ("Provider Needs to be set ...") raise Exception ("Provider Needs to be set ...")
def get_default_port(self) : def get_default_port(self) :
@ -122,6 +123,8 @@ class BaseReader(SQLBase):
sql = _args['sql'] sql = _args['sql']
else: else:
_table = _args['table'] if 'table' in _args else self._table _table = _args['table'] if 'table' in _args else self._table
if self._schema and type(self._schema) == str :
_table = f'{self._schema}.{_table}'
sql = f'SELECT * FROM {_table}' sql = f'SELECT * FROM {_table}'
return self.apply(sql) return self.apply(sql)
@ -132,6 +135,7 @@ class BaseWriter (SQLBase):
""" """
def __init__(self,**_args): def __init__(self,**_args):
super().__init__(**_args) super().__init__(**_args)
def write(self,_data,**_args): def write(self,_data,**_args):
if type(_data) == dict : if type(_data) == dict :
_df = pd.DataFrame(_data) _df = pd.DataFrame(_data)
@ -151,5 +155,8 @@ class BaseWriter (SQLBase):
# _mode['schema'] = _args['schema'] # _mode['schema'] = _args['schema']
# if 'if_exists' in _args : # if 'if_exists' in _args :
# _mode['if_exists'] = _args['if_exists'] # _mode['if_exists'] = _args['if_exists']
if 'schema' in _args and type(_args['schema']) == str:
self._schema = _args.get('schema',None)
if self._schema :
_mode['schema'] = self._schema
_df.to_sql(_table,self._engine,**_mode) _df.to_sql(_table,self._engine,**_mode)