Feat (system): new WrkActJob

app/iLibrary/src/iLibrary/Libr/getInfoForLibrary.py

@@ -2,7 +2,7 @@
 
 
 
-class getInfoForLibrary:
+class GetInfoForLibrary:
     def __init__(self, connection, mapepire=False):
         self.conn = connection
         self.mapepire = mapepire

app/iLibrary/src/iLibrary/Libr/saveLibrary.py

@@ -4,7 +4,7 @@
 from ..util_functions.helper import create_success_envelope, create_error_envelope
 
 
-class saveLibrary:
+class SaveLibrary:
     def __init__(self, connection, mapepire=False):
         """
         Initializes the saveLibrary parent class.

app/iLibrary/src/iLibrary/Library.py

@@ -5,7 +5,7 @@
 
 
 
-class Library(getInfoForLibrary, saveLibrary):
+class Library(GetInfoForLibrary, SaveLibrary):
     """
     A class to manage libraries and files on an IBM i system.
 

app/iLibrary/src/iLibrary/System/getWrkactjob.py

@@ -0,0 +1,93 @@
+from ..util_functions.helper import create_success_envelope, create_error_envelope
+
+class GetWrkActJob:
+    def __init__(self, connection, mapepire=False):
+        self.conn = connection
+        self.mapepire = mapepire
+    """
+    Handles user information retrieval and messaging functionalities.
+
+    This class provides methods to interact with the database for retrieving user information
+    and to send messages to specified users. It supports data retrieval in different formats
+    (e.g., JSON or tuple), and it enables system messaging with configurable options.
+
+    :ivar conn: Database connection object used for executing queries.
+    :type conn: Any
+    """
+    def get_active_jobs(self) -> dict[str, str]:
+        """
+        Retrieves information about active jobs from the system.
+
+        This method queries the database for active job information using the
+        QSYS2.ACTIVE_JOB_INFO() table function. It processes the retrieved data and
+        returns it in a formatted envelope. If no active jobs are found, or an error
+        occurs, an appropriate error envelope is returned.
+
+        Returns:
+            dict[str, str]: A dictionary representing a success or error envelope.
+
+        Raises:
+            Exception: If an error occurs during the database query or data processing.
+        """
+        sql_query = "SELECT * FROM TABLE(QSYS2.ACTIVE_JOB_INFO())"
+
+
+        try:
+            with self.conn.cursor() as cursor:
+                cursor.execute(sql_query)
+                rows = cursor.fetchall()
+                if self.mapepire:
+                    data = rows.get('data', []) if isinstance(rows, dict) else rows
+                    return create_success_envelope(data)
+                if not rows:
+                    error_msg = f"No active jobs found"
+                    return create_error_envelope(error_msg, func_name="getwrkactjob")
+
+                # Get column names
+                columns = [column[0] for column in cursor.description]
+
+
+                results = [dict(zip(columns, r)) for r in rows]
+                return create_success_envelope(results)
+
+
+
+        except Exception as e:
+            return create_error_envelope(error_msg=str(e), func_name="getwrkactjob")
+
+    def get_active_jobs_filter_by_subsystem(self, subsystem:str) -> dict[str, str]:
+        """
+        Retrieves information about active jobs from the system.
+        This method queries the database for active job information using the
+        """
+
+        if not isinstance(subsystem, str):
+            raise TypeError(f"Parameter 'subsystem' must be a str, got '{type(subsystem).__name__}'")
+
+        sql_query = f"""
+                    SELECT *
+                    FROM TABLE(QSYS2.ACTIVE_JOB_INFO(SUBSYSTEM_LIST_FILTER => ?)) 
+                    """
+        try:
+            with self.conn.cursor() as cursor:
+                cursor.execute(sql_query, (f'{subsystem}',))
+                rows = cursor.fetchall()
+                if self.mapepire:
+                    data = rows.get('data', []) if isinstance(rows, dict) else rows
+                    return create_success_envelope(data)
+                if not rows:
+                    error_msg = f"No active jobs found"
+                    return create_error_envelope(error_msg, func_name="get_ActiveJob_filter_by_subsystem")
+
+                # Get column names
+                columns = [column[0] for column in cursor.description]
+
+
+                results = [dict(zip(columns, r)) for r in rows]
+                return create_success_envelope(results)
+
+
+
+        except Exception as e:
+            return create_error_envelope(error_msg=str(e), func_name="get_ActiveJob_filter_by_subsystem")
+

app/iLibrary/src/iLibrary/User.py

@@ -3,7 +3,7 @@
 from .Usr.getUserInfoForUser import *
 from .Usr.sendMSG import *
 
-class User(getUserInfoForUser, sendMSG):
+class User(GetUserInfoForUser, sendMSG):
     """
         A class to manage User on IBMi System
 

app/iLibrary/src/iLibrary/Usr/getUserInfoForUser.py

@@ -1,7 +1,7 @@
 import pyodbc
 from ..util_functions.helper import create_success_envelope, create_error_envelope
 
-class getUserInfoForUser():
+class GetUserInfoForUser():
     def __init__(self, connection, mapepire=False):
         self.conn = connection
         self.mapepire = mapepire

app/iLibrary/src/iLibrary/__init__.py

@@ -1,3 +1,4 @@
 from .Library import Library
 from .User import User
-from .IFS import IFS
+from .IFS import IFS
+from .system import System

app/iLibrary/src/iLibrary/ifs/ifs_logic.py

@@ -50,4 +50,4 @@ def readIFS(self, path_to_read:str, subtrees:bool=True) -> dict[str, str]:
 
 
         except Exception as e:
-            return create_error_envelope(error_msg=str(e), func_name="readIFS")
+            return create_error_envelope(error_msg=str(e), func_name="readIFS")

app/iLibrary/src/iLibrary/system.py

@@ -0,0 +1,93 @@
+from mapepire_python import connect
+import pyodbc
+from .System.getWrkactjob import *
+
+class System(GetWrkActJob):
+    """
+    A class to manage the System on an IBM i system.
+
+    """
+
+    # ------------------------------------------------------
+    # __init__ - initzialise the class
+    # ------------------------------------------------------
+    def __init__(self, db_user: str, db_password: str, db_host: str, db_driver: str, mapepire: bool = False):
+        """
+        Initializes the class attributes for a database connection.
+        The actual connection is established in the __enter__ method.
+
+        Args:
+            db_user (str): The user ID for the database connection.
+            db_password (str): The password for the database user.
+            db_host (str): The system/host name for the database connection.
+            db_driver (str): The ODBC driver to be used.
+        """
+        self.db_user = db_user
+        self.db_host = db_host
+        self.db_driver = db_driver
+        self.db_password = db_password
+        self.mapepire = mapepire
+
+    # ------------------------------------------------------
+    # __enter__ - enter to the class
+    # ------------------------------------------------------
+    def __enter__(self) -> 'System':
+        """
+        Establishes the database connection when entering a 'with' block.
+        """
+        try:
+            if not self.mapepire:
+                conn_str = (
+                    f"DRIVER={self.db_driver};"
+                    f"SYSTEM={self.db_host};"
+                    f"UID={self.db_user};"
+                    f"PWD={self.db_password};"
+                )
+                self.conn = pyodbc.connect(conn_str, autocommit=True)
+
+            else:
+                conn_str = {
+                      "host": self.db_host,
+                      "port": 8076,
+                      "user": self.db_user,
+                      "password": self.db_password,
+                    }
+                self.conn = connect(conn_str)
+            super().__init__(self.conn, mapepire=self.mapepire)
+            return self
+
+        except pyodbc.Error as ex:
+            sqlstate = ex.args[0]
+            print(f"Database connection failed with error: {sqlstate}")
+            raise
+        except Exception as e:
+
+            print(f"Database connection failed with error: {e}")
+            raise
+
+    # ------------------------------------------------------
+    # __exit__ - leave the class
+    # ------------------------------------------------------
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """
+        Closes the database connection when exiting a 'with' block.
+        This method is called automatically, even if an error occurred.
+        """
+        self.iclose()
+
+
+    # ------------------------------------------------------
+    # iClose - close connection
+    # ------------------------------------------------------
+    def iclose(self):
+        if not self.conn:
+            return
+
+        try:
+            # Both pyodbc and mapepire-python support .close()
+            # but mapepire MUST have it called to kill background threads
+            self.conn.close()
+        except Exception:
+            pass
+        finally:
+            self.conn = None

app/iLibrary/tests/test_user.py

@@ -55,15 +55,15 @@ def test_get_single_user_info_mapepire(mock_user_context):
         data = lib.getSingleUserInformation(username=TEST_USER)
 
     # --- Assertions ---
-    # 1. Assert constructor received the correct arguments
+
     mock_user_class.assert_called_once_with(
         DB_USER, DB_PASSWORD, DB_SYSTEM, DB_DRIVER, mapepire=True
     )
 
-    # 2. Assert the method was called with the correct username
+
     user_instance.getSingleUserInformation.assert_called_once_with(username=TEST_USER)
 
-    # 3. Assert the data returned matches the input (ALBEER == ALBEER)
+
     assert data["USERNAME"] == TEST_USER
 
 

dev_test.py

@@ -2,7 +2,7 @@
 from os.path import join, dirname
 import os
 from dotenv import load_dotenv
-from iLibrary import Library, User, IFS
+from iLibrary import Library, User, IFS, System
 from os.path import dirname
 
 #load ENV file and get the Connection Settings
@@ -13,6 +13,31 @@
 DB_PASSWORD = os.environ.get("DB_PASSWORD")
 DB_SYSTEM = os.environ.get("DB_SYSTEM")
 
+def getSingleLibraryInfo():
+    USE_MAPEPIRE = False
+
+    try:
+        # Establish a connection to the IBM i system using the User class
+        # The context manager ensures the connection is properly opened and closed
+        with System(DB_USER, DB_PASSWORD, DB_SYSTEM, DB_DRIVER, mapepire=USE_MAPEPIRE) as u:
+
+            # Call the method to retrieve all users from the system
+            # The result is returned as a JSON string
+            raw_result = u.get_active_jobs()
+
+            # Parse the JSON string into a Python object (list/dictionary)
+            data = json.loads(raw_result)
+            counter = data['metadata'].get('count')
+
+            # Pretty-print the parsed data with indentation for readability
+            print(json.dumps(data, indent=4))
+            print(counter)
+    # Handle any exceptions that occur during connection or data retrieval
+    except Exception as e:
+        # Print the error message for debugging
+        print(e)
 
 if __name__ == "__main__":
-    print('Nothing to do')
+    getSingleLibraryInfo()
+
+