Rewrote pr_list_database.py to use wxflow's SQLiteDB Class

ci/scripts/pr_list_database.py

@@ -2,154 +2,126 @@
 
 import sys
 import os
-from pathlib import Path
-from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter, REMAINDER, ZERO_OR_MORE
-import sqlite3
+from wxflow import SQLiteDB, SQLiteDBError
+from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter, REMAINDER
 
 
 def full_path(string):
     """
-    Gets the absolute path of the given file and confirms the directory exists
+    full_path Get the absolute path of a file or directory.
 
     Parameters
     ----------
     string : str
-        Path to a file
+        The relative path of the file or directory.
 
     Returns
-    --------
+    -------
     str
-        Absolute path of input path
+        The absolute path of the file or directory.
 
     Raises
-    -------
+    ------
     NotADirectoryError
-        If the target directory for the file does not exist.
-
+        If the provided string does not represent a valid file or directory.
     """
-
     if os.path.isfile(string) or os.path.isdir(os.path.dirname(string)):
         return os.path.abspath(string)
     else:
         raise NotADirectoryError(string)
 
 
-def sql_connection(filename: os.path) -> sqlite3.Connection:
+def create_table(db: SQLiteDB):
     """
-    Returns an Sqlite3 Cursor object from a given path to a sqlite3 database file
+    Create a new table in a database.
 
     Parameters
     ----------
-    filename : Path
-        Full path to a sqlite3 database file
-
-    Returns
-    -------
-    sqlite3.Connection
-        Sqlite3 Connection object for updating table
-
+    db : SQLiteDB
+        The database to create.
     """
-    try:
-        return sqlite3.connect(filename)
-    except sqlite3.Error:
-        print(sqlite3.Error)
-        sys.exit(-1)
+    db.create_table('pr_list', ['pr INTEGER PRIMARY KEY UNIQUE', 'state TEXT', 'status TEXT', 'reset_id INTEGER', 'cases TEXT'])
 
 
-def sql_table(obj: sqlite3.Cursor) -> None:
+def add_pr(db: SQLiteDB, pr: str):
     """
-    Creates the initial sqlite3 table for PR states and status
+    Add a pull request to the database.
 
     Parameters
     ----------
-    obj : sqlite3.Cursor
-         Cursor object for Sqlite3
-
-    """
-
-    obj.execute("CREATE TABLE processing(pr integer PRIMARY KEY, state text, status text, reset_id integer, cases text)")
-
-
-def sql_insert(obj: sqlite3.Cursor, entities: list) -> None:
+    ci_database : SQLiteDB
+        The database to add the pull request to.
+    pr : str
+        The pull request to add.
     """
-    Inserts a new row in sqlite3 table with PR, state, and status
+    entities = (pr, 'Open', 'Ready', 0, 'ci_repo')
+    try:
+        db.insert_data('pr_list', entities)
+    except (SQLiteDBError.IntegrityError) as e:
+        if 'unique' in str(e).lower():
+            print(f"pr {pr} already is in list: nothing added")
 
-    Parameters
-    ----------
-    obj : sqlite3.Cursor
-        Cursor object for Sqlite3
-    entities : list
-        A list of four string values that go into sqlite table (pr, state, status, reset_id, cases)
-            pr: pull request number
-            state: The new value for the state (Open, Closed)
-            status: The new value for the status (Ready, Running, Failed)
-            reset_id: The value for number of times reset_id to Ready
-            cases: String containing case selection information
 
+def update_pr(db: SQLiteDB, args):
     """
-
-    obj.execute('INSERT INTO processing(pr, state, status, reset_id, cases) VALUES(?, ?, ?, ?, ?)', entities)
-
-
-def sql_update(obj: sqlite3.Cursor, pr: str, updates: dict) -> None:
-    """Updates table for a given pr with new values for state and status
+    Update a pull request in the database.
 
     Parameters
     ----------
-    obj : sqlite.sql_connection
-        sqlite3 Cursor Object
-    pr : str
-        The given pr number to update in the table
-    updates : dict
-        Dictionary of values to update for a given PR to include by postion
-        state, The new value for the state (Open, Closed)
-        status, The new value for the status (Ready, Running, Failed)
-        reset_id, The value for number of times reset_id to Ready
-        cases, Information regarding which cases are used (i.e. self PR)
-
+    db : SQLiteDB
+        The database to update the pull request in.
+    args : argparse.Namespace
+        The command line arguments.
     """
+    if len(args.update_pr) < 2:
+        print(f"update_pr must have at least one vaule to update")
+        sys.exit(0)
 
     update_list = ['state', 'status', 'reset_id', 'cases']
-    rows = sql_fetch(obj)
-    for value in updates:
+    for value in args.update_pr[1:]:
         update = update_list.pop(0)
-        obj.execute(f'UPDATE processing SET "{update}" = "{value}" WHERE pr = {pr}')
-
+        db.update_data('pr_list', update, value, 'pr', args.update_pr[0])
 
-def sql_fetch(obj: sqlite3.Cursor) -> list:
-    """ Gets list of all rows in table
-
-    Parameters
-    ----------
-    obj : sqlite.sql_connection
-        sqlite3 Cursor Object
 
+def display_db(db, display):
     """
-
-    obj.execute('SELECT * FROM processing')
-    return obj.fetchall()
-
-
-def sql_remove(obj: sqlite3.Cursor, pr: str) -> None:
-    """ Removes the row from table with given pr number
+    Display the database.
 
     Parameters
     ----------
-    obj : sqlite.sql_connection
-        sqlite3 Connection Object
-    pr : str
-        pr number acting as key for removing the row with in it
+    ci_database : SQLiteDB
+        The database to display.
+    display : list
+        The command line argument values.
 
+    Returns
+    -------
+    list
+        The rows of the database.
     """
+    values = []
+    if len(display) == 1:
+        rows = db.fetch_data('pr_list', ['pr', 'state', 'status', 'reset_id', 'cases'], f'pr = {display[0]}')
+    else:
+        rows = db.fetch_data('pr_list', ['pr', 'state', 'status', 'reset_id', 'cases'])
+    for row in rows:
+        values.append(' '.join(map(str, row)))
 
-    obj.execute(f'DELETE FROM processing WHERE pr = {pr}').rowcount
+    return values
 
 
 def input_args():
+    """
+    Parse command line arguments.
 
-    description = """Arguments for creating and updating db file for pr states
+    Returns
+    -------
+    argparse.Namespace
+        The parsed command line arguments.
     """
 
+    description = """Arguments for creating and updating db file for pr states
+    """
     parser = ArgumentParser(description=description,
                             formatter_class=ArgumentDefaultsHelpFormatter)
 
@@ -160,7 +132,6 @@ def input_args():
     parser.add_argument('--update_pr', nargs=REMAINDER, metavar=('pr', 'state', 'status', 'reset_id', 'cases'),
                         help='updates state and status of a given pr', required=False)
     parser.add_argument('--display', nargs='*', help='output pr table', required=False)
-
     args = parser.parse_args()
     return args
 
@@ -174,42 +145,19 @@ def input_args():
             print(f'Error: {args.dbfile} does not exsist')
             sys.exit(-1)
 
-    con = sql_connection(args.dbfile)
-    obj = con.cursor()
+    ci_database = SQLiteDB(args.dbfile)
+    ci_database.connect()
 
     if args.create:
-        sql_table(obj)
-
+        create_table(ci_database)
     if args.add_pr:
-        rows = sql_fetch(obj)
-        for row in rows:
-            if str(row[0]) == str(args.add_pr[0]):
-                print(f"pr {row[0]} already is in list: nothing added")
-                sys.exit(0)
-
-        entities = (args.add_pr[0], 'Open', 'Ready', 0, 'ci_repo')
-        sql_insert(obj, entities)
-
+        add_pr(ci_database, args.add_pr[0])
     if args.update_pr:
-        if len(args.update_pr) < 2:
-            print(f"update_pr must have at least one vaule to update")
-            sys.exit(0)
-        pr = args.update_pr[0]
-
-        sql_update(obj, pr, args.update_pr[1:])
-
+        update_pr(ci_database, args)
     if args.remove_pr:
-        sql_remove(obj, args.remove_pr[0])
-
+        ci_database.remove_data('pr_list', 'PR', args.remove_pr[0])
     if args.display is not None:
-        rows = sql_fetch(obj)
-        if len(args.display) == 1:
-            for row in rows:
-                if int(args.display[0]) == int(row[0]):
-                    print(' '.join(map(str, row)))
-        else:
-            for row in rows:
-                print(' '.join(map(str, row)))
-
-    con.commit()
-    con.close()
+        for rows in display_db(ci_database, args.display):
+            print(rows)
+
+    ci_database.disconnect()