Merge pull request #15 from lnyemba/v2.2.0

V2.2.0 - Upgraded to shared collaborative environments, with duckdb support

README.md

@@ -8,7 +8,7 @@ Mostly data scientists that don't really care about the underlying database and
 
 1. Familiarity with **pandas data-frames**
 2. Connectivity **drivers** are included
-3. Mining data from various sources
+3. Reading/Writing data from various sources
 4. Useful for data migrations or **ETL**
 
 
@@ -19,6 +19,15 @@ Within the virtual environment perform the following :
     pip install git+https://github.com/lnyemba/data-transport.git
 
 
+## What's new
+
+Unlike older versions 2.0 and under, we focus on collaborative environments like jupyter-x servers; apache zeppelin:
+
+    1. Simpler syntax to create reader or writer
+    2. auth-file registry that can be referenced using a label
+    3. duckdb support
+
+
 ## Learn More
 
-We have available notebooks with sample code to read/write against mongodb, couchdb, Netezza, PostgreSQL, Google Bigquery, Databricks, Microsoft SQL Server, MySQL ... Visit [data-transport homepage](https://healthcareio.the-phi.com/data-transport)
+We have available notebooks with sample code to read/write against mongodb, couchdb, Netezza, PostgreSQL, Google Bigquery, Databricks, Microsoft SQL Server, MySQL ... Visit [data-transport homepage](https://healthcareio.the-phi.com/data-transport)

bin/transport

@@ -13,29 +13,6 @@ The above copyright notice and this permission notice shall be included in all c
 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
-Usage :
-	transport help -- will print this page
-
-    transport move <path> [index]
-        <path>  path to the configuration file
-        <index> optional index within the configuration file
-
-e.g: configuration file (JSON formatted)
-    - single source to a single target
-
-    {"source":{"provider":"http","url":"https://cdn.wsform.com/wp-content/uploads/2020/06/agreement.csv"}
-        "target":{"provider":"sqlite3","path":"transport-demo.sqlite","table":"agreement"}
-    }
-
-    - single source to multiple targets
-    {
-        "source":{"provider":"http","url":"https://cdn.wsform.com/wp-content/uploads/2020/06/agreement.csv"},
-        "target":[
-            {"provider":"sqlite3","path":"transport-demo.sqlite","table":"agreement},
-            {"provider":"mongodb","db":"transport-demo","collection":"agreement"}
-        ]
-    }
-
 """
 import pandas as pd
 import numpy as np
@@ -53,9 +30,13 @@ import typer
 from typing_extensions import Annotated
 from typing import Optional
 import time
+from termcolor import colored
 
 app = typer.Typer()
-
+REGISTRY_PATH=os.sep.join([os.environ['HOME'],'.data-transport'])
+REGISTRY_FILE= 'transport-registry.json'
+CHECK_MARK = ' '.join(['[',colored(u'\u2713', 'green'),']'])
+TIMES_MARK= ' '.join(['[',colored(u'\u2717','red'),']'])
 # @app.command()
 def help() :     
 	print (__doc__)
@@ -66,9 +47,9 @@ def wait(jobs):
 
 @app.command(name="apply")
 def apply (path:Annotated[str,typer.Argument(help="path of the configuration file")],
-        index:int = typer.Option(help="index of the item of interest, otherwise everything in the file will be processed")):
+        index:int = typer.Option(default= None, help="index of the item of interest, otherwise everything in the file will be processed")):
     """
-    This function applies data transport from one source to one or several others
+    This function applies data transport ETL feature to read data from one source to write it one or several others
     """
     # _proxy = lambda _object: _object.write(_object.read())
     if os.path.exists(path):
@@ -111,48 +92,60 @@ def version():
 
 @app.command()
 def generate (path:Annotated[str,typer.Argument(help="path of the ETL configuration file template (name included)")]):
-	"""
+    """
-	This function will generate a configuration template to give a sense of how to create one
+    This function will generate a configuration template to give a sense of how to create one
-	"""
+    """
-	_config = [
+    _config = [
-          {
+            {
-               "source":{"provider":"http","url":"https://raw.githubusercontent.com/codeforamerica/ohana-api/master/data/sample-csv/addresses.csv"},
+                "source":{"provider":"http","url":"https://raw.githubusercontent.com/codeforamerica/ohana-api/master/data/sample-csv/addresses.csv"},
-             "target":
+                "target":
             [{"provider":"files","path":"addresses.csv","delimiter":","},{"provider":"sqlite","database":"sample.db3","table":"addresses"}]
             }
             ]
-	file = open(path,'w')
+    file = open(path,'w')
-	file.write(json.dumps(_config))
+    file.write(json.dumps(_config))
-	file.close()
+    file.close()
+    print (f"""{CHECK_MARK} Successfully generated a template ETL file at {path}""" )
+    print ("""NOTE: Each line (source or target) is the content of an auth-file""")
+
+
+
+@app.command(name="init")
+def initregistry (email:Annotated[str,typer.Argument(help="email")],
+                  path:str=typer.Option(default=REGISTRY_PATH,help="path or location of the configuration file"), 
+                  override:bool=typer.Option(default=False,help="override existing configuration or not")):
+    """
+    This functiion will initialize the registry and have both application and calling code loading the database parameters by a label
+
+    """
+    try:
+        transport.registry.init(email=email, path=path, override=override)
+        _msg = f"""{CHECK_MARK} Successfully wrote configuration to {path} from {email}"""
+    except Exception as e:
+        _msg = f"{TIMES_MARK} {e}"
+    print (_msg)
+    print ()
+@app.command(name="register")
+def register (label:Annotated[str,typer.Argument(help="unique label that will be used to load the parameters of the database")],
+              auth_file:Annotated[str,typer.Argument(help="path of the auth_file")],
+              default:bool=typer.Option(default=False,help="set the auth_file as default"),
+              path:str=typer.Option(default=REGISTRY_PATH,help="path of the data-transport registry file")):
+    """
+    This function will register an auth-file i.e database connection and assign it a label, 
+    Learn more about auth-file at https://healthcareio.the-phi.com/data-transport
+    """
+    try:
+        if transport.registry.exists(path) :
+            transport.registry.set(label=label,auth_file=auth_file, default=default, path=path)
+            _msg = f"""{CHECK_MARK} Successfully added label "{label}" to data-transport registry"""
+        else:
+            _msg = f"""{TIMES_MARK} Registry is not initialized, please initialize the registry (check help)"""
+    except Exception as e:
+        _msg = f"""{TIMES_MARK} {e}"""
+    print (_msg)
+    
+    pass
 if __name__ == '__main__' :
      app()
-# 	#
+	
-# 	# Load information from the file ...
-# 	if 'help' in SYS_ARGS :
-# 		print (__doc__)
-# 	else:
-# 		try:
-# 			_info = json.loads(open(SYS_ARGS['config']).read())
-# 			if 'index' in SYS_ARGS :
-# 				_index = int(SYS_ARGS['index'])
-# 				_info = [_item for _item in _info if _info.index(_item) == _index]
-# 				pass
-# 			elif 'id' in SYS_ARGS :
-# 				_info = [_item for _item in _info if 'id' in _item and _item['id'] == SYS_ARGS['id']]
-			
-# 			procs = 1 if 'procs' not in SYS_ARGS else int(SYS_ARGS['procs'])
-# 			jobs = transport.factory.instance(provider='etl',info=_info,procs=procs)
-# 			print ([len(jobs),' Jobs are running'])
-# 			N = len(jobs)
-# 			while jobs :
-# 				x = len(jobs)
-# 				jobs = [_job for _job in jobs if _job.is_alive()]
-# 				if x != len(jobs) :
-# 					print ([len(jobs),'... jobs still running'])
-# 				time.sleep(1)
-# 			print ([N,' Finished running'])
-# 		except Exception as e:
-			
-# 			print (e)
-			
 	

info/__init__.py

@@ -1,6 +1,6 @@
 __app_name__  = 'data-transport'
 __author__ = 'The Phi Technology'
-__version__= '2.0.4'
+__version__= '2.2.0'
 __email__  = "info@the-phi.com"
 __license__=f"""
 Copyright 2010 - 2024, Steve L. Nyemba
@@ -12,3 +12,10 @@ The above copyright notice and this permission notice shall be included in all c
 THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 """
+
+__whatsnew__=f"""version {__version__}, focuses on collaborative environments like jupyter-base servers (apache zeppelin; jupyter notebook, jupyterlab, jupyterhub)
+
+    1. simpler syntax to create readers/writers
+    2. auth-file registry that can be referenced using a label
+    3. duckdb support
+"""

setup.py

@@ -18,8 +18,8 @@ args    = {
     # "packages":["transport","info","transport/sql"]},
 
     "packages": find_packages(include=['info','transport', 'transport.*'])}
-args["keywords"]=['mongodb','couchdb','rabbitmq','file','read','write','s3','sqlite']
+args["keywords"]=['mongodb','duckdb','couchdb','rabbitmq','file','read','write','s3','sqlite']
-args["install_requires"] = ['pyncclient','pymongo','sqlalchemy','pandas','typer','pandas-gbq','numpy','cloudant','pika','nzpy','boto3','boto','pyarrow','google-cloud-bigquery','google-cloud-bigquery-storage','flask-session','smart_open','botocore','psycopg2-binary','mysql-connector-python','numpy','pymssql']
+args["install_requires"] = ['pyncclient','duckdb-engine','pymongo','sqlalchemy','pandas','typer','pandas-gbq','numpy','cloudant','pika','nzpy','boto3','boto','pyarrow','google-cloud-bigquery','google-cloud-bigquery-storage','flask-session','smart_open','botocore','psycopg2-binary','mysql-connector-python','numpy','pymssql']
 args["url"] =   "https://healthcareio.the-phi.com/git/code/transport.git"
 args['scripts'] = ['bin/transport']
 # if sys.version_info[0] == 2 :

transport/__init__.py

@@ -22,12 +22,15 @@ from transport import sql, nosql, cloud, other
 import pandas as pd
 import json
 import os
-from info import __version__,__author__,__email__,__license__,__app_name__
+from info import __version__,__author__,__email__,__license__,__app_name__,__whatsnew__
 from transport.iowrapper import IWriter, IReader, IETL
 from transport.plugins import PluginLoader
 from transport import providers
+import copy 
+from transport import registry
 
 PROVIDERS = {}
+
 def init():
     global PROVIDERS
     for _module in [cloud,sql,nosql,other] :
@@ -35,7 +38,11 @@ def init():
             if _provider_name.startswith('__') or _provider_name == 'common':
                 continue
             PROVIDERS[_provider_name] = {'module':getattr(_module,_provider_name),'type':_module.__name__}
-
+def _getauthfile (path) :
+    f = open(path)
+    _object = json.loads(f.read())
+    f.close()
+    return _object
 def instance (**_args):
     """
     This function returns an object of to read or write from a supported database provider/vendor
@@ -45,6 +52,16 @@ def instance (**_args):
     kwargs      These are arguments that are provider/vendor specific
     """
     global PROVIDERS
+    # if not registry.isloaded () :
+    #     if ('path' in _args and registry.exists(_args['path'] )) or registry.exists():
+    #         registry.load() if 'path' not in _args else registry.load(_args['path'])
+    #         print ([' GOT IT'])
+    # if 'label' in _args and registry.isloaded():
+    #     _info = registry.get(_args['label'])
+    #     if _info :
+    #         #
+    #         _args = dict(_args,**_info)
+
     if 'auth_file' in _args:
         if os.path.exists(_args['auth_file']) :
             #
@@ -60,6 +77,20 @@ def instance (**_args):
         else:
             filename = _args['auth_file']
             raise Exception(f" {filename} was not found or is invalid")
+    if 'provider' not in _args and 'auth_file' not in _args :
+        if not registry.isloaded () :
+            if ('path' in _args and registry.exists(_args['path'] )) or registry.exists():
+                registry.load() if 'path' not in _args else registry.load(_args['path'])
+        _info = {}
+        if 'label' in _args and registry.isloaded():
+            _info = registry.get(_args['label'])
+        else:
+            _info = registry.get()    
+        if _info :
+            #
+            # _args = dict(_args,**_info)
+            _args = dict(_info,**_args) #-- we can override the registry parameters with our own arguments
+
     if 'provider' in _args and _args['provider'] in PROVIDERS :
         _info = PROVIDERS[_args['provider']]
         _module = _info['module']
@@ -103,6 +134,8 @@ class get :
     """
     @staticmethod
     def reader (**_args):
+        if not _args or 'provider' not in _args:
+            _args['label'] = 'default'
         _args['context'] = 'read'
         return instance(**_args)
     @staticmethod
@@ -110,6 +143,8 @@ class get :
         """
         This function is a wrapper that will return a writer to a database. It disambiguates the interface
         """
+        if not _args :
+            _args['label'] = 'default'
         _args['context'] = 'write'
         return instance(**_args)
     @staticmethod

transport/iowrapper.py

@@ -52,6 +52,13 @@ class IO:
         if hasattr(self._agent,'apply') :
             return self._agent.apply(_query)
         return None
+    def submit(self,_query):
+        return self.delegate('submit',_query)
+    def delegate(self,_name,_query):
+        if hasattr(self._agent,_name) :
+            pointer = getattr(self._agent,_name)
+            return pointer(_query)
+        return None
 class IReader(IO):
     """
     This is a wrapper for read functionalities

transport/other/callback.py

@@ -1,3 +1,7 @@
+"""
+This module uses callback architectural style as a writer to enable user-defined code to handle the output of a reader
+The intent is to allow users to have control over the output of data to handle things like logging, encryption/decryption and other
+"""
 import queue
 from threading import Thread, Lock
 # from transport.common import Reader,Writer

transport/other/console.py

@@ -1,3 +1,6 @@
+"""
+This class uses classback pattern to allow output to be printed to the console (debugging)
+"""
 from . import callback
 
 

transport/providers/__init__.py

@@ -10,8 +10,11 @@ HTTP='http'
 BIGQUERY	='bigquery'
 FILE 	= 'file'
 ETL = 'etl'
+
 SQLITE = 'sqlite'
 SQLITE3= 'sqlite3'
+DUCKDB = 'duckdb'
+
 REDSHIFT = 'redshift'
 NETEZZA = 'netezza'
 MYSQL = 'mysql'
@@ -42,5 +45,6 @@ PGSQL	= POSTGRESQL
 AWS_S3  = 's3'
 RABBIT = RABBITMQ
 
+
 # QLISTENER = 'qlistener'
     

transport/registry.py

@@ -0,0 +1,102 @@
+import os
+import json
+from info import __version__
+import copy
+import transport
+
+"""
+This class manages data from the registry and allows (read only)
+@TODO: add property to the DATA attribute
+"""
+
+REGISTRY_PATH=os.sep.join([os.environ['HOME'],'.data-transport'])
+#
+# This path can be overriden by an environment variable ...
+#
+if 'DATA_TRANSPORT_REGISTRY_PATH' in os.environ :
+    REGISTRY_PATH = os.environ['DATA_TRANSPORT_REGISTRY_PATH']
+REGISTRY_FILE= 'transport-registry.json'
+
+DATA = {}
+
+def isloaded ():
+    return DATA not in [{},None]
+def exists (path=REGISTRY_PATH) :
+    """
+    This function determines if there is a registry at all
+    """
+    p = os.path.exists(path)
+    q = os.path.exists( os.sep.join([path,REGISTRY_FILE]))
+    
+    return p and q
+def load (_path=REGISTRY_PATH):
+    global DATA
+    
+    if exists(_path) :
+        path = os.sep.join([_path,REGISTRY_FILE])
+        f = open(path)
+        DATA = json.loads(f.read())
+        f.close()
+def init (email,path=REGISTRY_PATH,override=False):
+    """
+    Initializing the registry and will raise an exception in the advent of an issue
+    """
+    p = '@' in email
+    q = False if '.' not in email else email.split('.')[-1] in ['edu','com','io','ai','org']
+    if p and q :
+        _config = {"email":email,'version':__version__}
+        if not os.path.exists(path):
+            os.makedirs(path)
+        filename = os.sep.join([path,REGISTRY_FILE])
+        if not os.path.exists(filename) or override == True :
+
+            f = open(filename,'w')
+            f.write( json.dumps(_config))
+            f.close()
+            # _msg = f"""{CHECK_MARK} Successfully wrote configuration to {path} from {email}"""
+            
+        else:
+            raise Exception (f"""Unable to write configuration, Please check parameters (or help) and try again""")
+    else:
+        raise Exception (f"""Invalid Input, {email} is not well formatted, provide an email with adequate format""")
+def lookup (label):
+    global DATA
+    return label in DATA
+def get (label='default') :
+    global DATA
+    return copy.copy(DATA[label]) if label in DATA else {}
+
+def set (label, auth_file, default=False,path=REGISTRY_PATH) :
+    """
+    This function will add a label (auth-file data) into the registry and can set it as the default
+    """
+    if label == 'default' :
+        raise Exception ("""Invalid label name provided, please change the label name and use the switch""")
+    reg_file = os.sep.join([path,REGISTRY_FILE])
+    if os.path.exists (auth_file) and os.path.exists(path) and os.path.exists(reg_file):
+        f = open(auth_file)
+        _info = json.loads(f.read())
+        f.close()
+        f = open(reg_file)
+        _config = json.loads(f.read())
+        f.close()
+
+        #
+        # set the proposed label
+        _object = transport.factory.instance(**_info)
+        if _object :
+            _config[label] = _info
+            if default :
+                _config['default'] = _info
+            #
+            # now we need to write this to the location
+            f = open(reg_file,'w')
+            f.write(json.dumps(_config))
+            f.close()
+        else:
+            raise Exception( f"""Unable to load file locate at {path},\nLearn how to generate auth-file with wizard found at https://healthcareio.the-phi.com/data-transport""")
+        pass         
+    else:
+        pass
+    pass
+

transport/sql/__init__.py

@@ -3,7 +3,7 @@ This namespace/package wrap the sql functionalities for a certain data-stores
     - netezza, postgresql, mysql and sqlite
     - mariadb, redshift (also included)
 """
-from . import postgresql, mysql, netezza, sqlite, sqlserver
+from . import postgresql, mysql, netezza, sqlite, sqlserver, duckdb
 
 
 #

transport/sql/duckdb.py

@@ -0,0 +1,21 @@
+"""
+This module implements the handler for duckdb (in memory or not)
+"""
+from transport.sql.common import Base, BaseReader, BaseWriter
+
+class Duck :
+    def __init__(self,**_args):
+        self.database = _args['database']
+    def get_provider(self):
+        return "duckdb"
+    
+    def _get_uri(self,**_args):
+        return f"""duckdb:///{self.database}"""
+class Reader(Duck,BaseReader) :
+    def __init__(self,**_args):
+        Duck.__init__(self,**_args)
+        BaseReader.__init__(self,**_args)
+class Writer(Duck,BaseWriter):
+    def __init__(self,**_args):
+        Duck.__init__(self,**_args)
+        BaseWriter.__init__(self,**_args)