diff --git a/bin/transport b/bin/transport index fd5d41b..4f9a7e8 100755 --- a/bin/transport +++ b/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 [index] - path to the configuration file - 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__) @@ -68,7 +49,7 @@ def wait(jobs): 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")): """ - 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): @@ -124,35 +105,40 @@ def generate (path:Annotated[str,typer.Argument(help="path of the ETL configurat file = open(path,'w') file.write(json.dumps(_config)) file.close() + +@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: + 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""" + 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) - + diff --git a/transport/__init__.py b/transport/__init__.py index d7d4518..2f97c0f 100644 --- a/transport/__init__.py +++ b/transport/__init__.py @@ -26,8 +26,11 @@ from info import __version__,__author__,__email__,__license__,__app_name__ 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] : @@ -45,6 +48,10 @@ def instance (**_args): kwargs These are arguments that are provider/vendor specific """ global PROVIDERS + if not registry.isloaded () : + registry.load() if 'path' not in _args else registry.load(_args['path']) + if 'label' in _args : + _info = registry.load(_args['label']) if 'auth_file' in _args: if os.path.exists(_args['auth_file']) : # diff --git a/transport/other/callback.py b/transport/other/callback.py index c56c175..aba2f02 100644 --- a/transport/other/callback.py +++ b/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 diff --git a/transport/other/console.py b/transport/other/console.py index 16f589a..b2f374b 100644 --- a/transport/other/console.py +++ b/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 diff --git a/transport/registry.py b/transport/registry.py new file mode 100644 index 0000000..9fe942d --- /dev/null +++ b/transport/registry.py @@ -0,0 +1,81 @@ +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']) +REGISTRY_FILE= 'transport-registry.json' + +DATA = {} + +def isloaded (): + return DATA not in [{},None] +def load (_path=REGISTRY_PATH): + global DATA + _path = os.sep.join([_path,REGISTRY_FILE]) + if os.path.exists(_path) : + 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'] + 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 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) : + 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: + _msg = 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 +