Update Code

Author: App Generator

util/__init__.py

@@ -0,0 +1,239 @@
+# -*- encoding: utf-8 -*-
+"""
+Copyright (c) 2019 - present AppSeed.us
+License: MIT
+"""
+
+from util.helpers import *
+
+from peewee import SqliteDatabase, MySQLDatabase, PostgresqlDatabase
+from playhouse.reflection import generate_models, print_model, print_table_sql
+from playhouse.dataset import DataSet
+
+def test():
+    print( 'working!' )
+
+class DbWrapper:
+
+    # private
+    _db     = None
+    _ds     = None
+    _models = None
+
+    # public data
+    driver  = None
+    db_name = None 
+    db_user = None 
+    db_pass = None 
+
+    # reset
+    def reset(self):
+
+        self._db     = None
+        self._ds     = None
+        self._models = None
+
+        self.driver  = None
+        self.db_name = None 
+        self.db_user = None 
+        self.db_pass = None 
+
+    # helpers
+    def connect(self):
+
+        if not self.driver:
+            print( ' > Error DB driver not set' )
+
+        if self.driver == COMMON.DB_SQLITE:
+            self._db = SqliteDatabase( self.db_name )
+            self._ds = DataSet( self._db )
+            return True
+        elif self.driver == COMMON.DB_MYSQL:
+            self._db = MySQLDatabase( self.db_name, user=self.db_user, password=self.db_pass )
+            self._ds = DataSet( self._db )
+            return True
+        elif self.driver == COMMON.DB_PGSQL:
+            self._db = PostgresqlDatabase( self.db_name, user=self.db_user, password=self.db_pass )
+            self._ds = DataSet( self._db )
+            return True
+        else:
+            print( ' > Error unsupported driver [' + driver + ']' )
+            return False
+
+    def load_models(self):
+
+        if not self._db:
+            print( ' > Error: No DB connection' )
+            return None
+
+        if not self._models:
+            self._models = generate_models( self._db )
+            return True
+
+    def get_model(self, aModelName):
+
+        if not self._db:
+            print( ' > Error: No DB connection' )
+            return None
+
+        if not self._models:
+            print( ' > Error: No DB models' )
+            return None
+
+        if not aModelName in self._models:
+            print( ' > Error Model [' + aModelName + '] not found' )
+            return None
+
+        return self._models[ aModelName ]
+
+    def get_model_data(self, aModelName):
+
+        aModel = self.get_model( aModelName )
+
+        if not aModel:
+            return None
+
+        return self._ds[ aModelName ].all()
+
+    def print_all_models(self):
+
+        if not self._db:
+            print( ' > Error: No DB connection' )
+            return
+
+        if not self._models:
+            print( ' > Error: No DB models' )
+            return
+
+        print( list(self._models.items()) )
+
+    def print_db_model(self, aModelName, aSql=False):
+
+        if not self._models:
+            print( ' > Error: No DB models' )
+            return
+
+        if not aModelName in self._models:
+            print( ' > Error Model [' + aModelName + '] not found' )
+            return
+
+        aModel = self._models[ aModelName ]
+
+        if aSql:
+            print_table_sql( aModel )
+        else:
+            print_model(aModel)
+
+        return
+
+    def sql_select(self, aModelName):
+
+        if not self._models:
+            print( ' > Error: No DB models' )
+            return
+
+        if not aModelName in self._models:
+            print( ' > Error Model [' + aModelName + '] not found' )
+            return None
+
+        aModel = self._models[ aModelName ]
+
+        for item in aModel.select():
+            print ( item )  
+
+    def dump_model_data(self, aModelName, aFileName=None):
+
+        items = self.get_model_data( aModelName )
+
+        if not items:
+            return None
+
+        if len( items ) == 0:
+            print( ' > Model [' + aModelName + '] is empty ' )
+            return None
+
+        header = items[0].keys()
+
+        sqlContent = ''
+        
+        for col in header:
+            sqlContent += col + ','
+
+        sqlContent += '\n'
+
+        for row in items:
+            
+            for i in row.values():
+                sqlContent += str( i ) + ','  
+            
+            sqlContent += '\n'
+
+        if not aFileName:
+            aFileName = get_date() + '_' + self.driver + '_' + aModelName
+
+        if not aFileName.endswith('.sql'):
+            aFileName += '.sql'
+
+        file_write('output/' + aFileName, sqlContent)
+        return True
+
+    def dump_tables(self, aFileName=None):
+
+        sqlContent = ''
+
+        if not self._models:
+            print( ' > Error: No DB models' )
+            return False
+
+        if not aFileName:
+            aFileName = get_date() + '_' + self.driver
+
+        if not aFileName.endswith('.sql'):
+            aFileName += '.sql'
+
+        for table_name in self._models.keys():       
+
+            # SQLite Engine Metadata
+            if 'sqlite_sequence' == table_name:
+                continue
+
+            # Unused
+            aModel = self._models[ table_name ]
+
+            # Hack the print
+            with Capturing() as output:
+
+                self.print_db_model( table_name, True)
+
+            sqlContent += '\n'    
+            sqlContent += '-- Table: ' + table_name + '\n'
+            sqlContent += h_list_to_s( output, '\n' )
+            sqlContent += ';\n'
+
+        file_write('output/' + aFileName, sqlContent)
+        return True
+
+    def dump_tables_data(self, aFileName=None):
+
+        sqlContent = ''
+
+        if not self._models:
+            print( ' > Error: No DB models' )
+            return False
+
+        if not aFileName:
+            aFileName = get_date() + '_' + self.driver
+
+        if not aFileName.endswith('.sql'):
+            aFileName += '.sql'
+
+        for table_name in self._models.keys():
+
+            # SQLite Engine Metadata
+            if 'sqlite_sequence' == table_name:
+                continue
+            
+            print( ' > Dump data for [' + table_name + ']' )
+            self.dump_model_data( table_name )
+
+        return True

util/helpers.py

@@ -0,0 +1,149 @@
+# -*- encoding: utf-8 -*-
+"""
+Copyright (c) 2019 - present AppSeed.us
+License: MIT
+"""
+
+from lib2to3.pgen2 import driver
+import os, csv, fnmatch, time, shutil
+
+from io import StringIO 
+import sys
+
+'''
+This file should contain primitives used in other modules
+'''
+
+class COMMON:
+    DB_SQLITE = 'SQLITE'
+    DB_MYSQL  = 'MYSQL'
+    DB_PGSQL  = 'PGSQL'
+
+def h_list_to_s( aList, aSeparator=',' ):
+    if isinstance(aList, list):
+        return aSeparator.join([str(elem) for elem in aList])
+    else: 
+        return aList
+
+def file_exists( aPath ):
+
+    try:
+
+        if open( aPath, 'r'):
+            return True
+
+    except:
+        return False   
+
+def file_p( aPath ):
+
+    try:
+
+        f = open( aPath, 'r') 
+        
+        if f:
+            return f
+
+        return None 
+    except:
+        return None   
+
+def file_read( aPath, as_list=False ):
+
+    try:
+
+        if not file_exists( aPath ):
+            return None
+
+        f = open( aPath, 'r')
+        if f:
+
+            if as_list:
+                content = f.readlines()
+            else:
+                content = f.read()    
+            
+            f.close()
+            return content
+
+    except UnicodeDecodeError as err:
+
+        return None
+
+    except:
+        return None
+
+def file_rm ( aPath ):
+    if file_exists( aPath ):
+        os.remove( aPath )
+
+def file_cp ( aSrc, aDest ):
+    if file_exists( aSrc ):
+        shutil.copyfile(aSrc, aDest)
+
+def file_write( aPath, aContent, isAppend=False ): 
+
+    try:
+
+        f = None
+
+        if file_exists( aPath ):
+            if isAppend:    
+                f = open( aPath, 'a+')
+            else:
+                f = open( aPath, 'w+')    
+        else:
+            f = open( aPath, 'w+')
+
+        if not f:
+            return False
+
+        f.seek(0) 
+        f.write( aContent )
+        f.truncate()
+        f.close()
+
+        return True
+
+    except IOError:
+        return False
+
+    except:
+        return False
+
+def list_files(dir_to_scan, ext='html'):
+
+    matches = []
+
+    for root, dirnames, filenames in os.walk( dir_to_scan ):
+        for filename in fnmatch.filter(filenames, '*.'+ ext):
+
+            item = os.path.join(root, filename)
+
+            #print ' **** type(item) = ' + str( type ( item ) )
+            matches.append( item )
+
+    return matches
+
+def list_sqlite( aPath ):
+
+    return list_files( aPath, 'sqlite3' )
+
+def get_tail( aPath ):
+    head, tail = os.path.split( aPath )
+    return tail
+
+def get_date(aFormat='%H_%M_%S'):
+    return time.strftime( aFormat, time.gmtime())
+
+# Credits:
+# https://stackoverflow.com/questions/16571150/how-to-capture-stdout-output-from-a-python-function-call
+class Capturing(list):
+    def __enter__(self):
+        self._stdout = sys.stdout
+        sys.stdout = self._stringio = StringIO()
+        return self
+    def __exit__(self, *args):
+        self.extend(self._stringio.getvalue().splitlines())
+        del self._stringio    # free up some memory
+        sys.stdout = self._stdout