Feature/add script helper

py_utils/log_config.py

@@ -0,0 +1,45 @@
+from datetime import datetime
+from . import utils
+from typing import List
+
+import logging
+import os
+import sys
+
+
+fmt = '%(asctime)s  %(levelname)-9s %(message)s [%(module)s:%(funcName)s:%(lineno)s]'
+logs_dir = "logs"
+todays_date = datetime.today().strftime('%Y-%m-%d')
+log_filename = os.path.join(logs_dir, f"log-{todays_date}.log")
+
+
+def configure_logging(handlers: List[logging.Handler] = []):
+    """Configures the root logger
+
+    Configures the root logger with default settings.
+
+    Args:
+        handlers (List[logging.Handler]): The handlers to add to the logger
+
+    Returns:
+        None
+    """
+    formatter = logging.Formatter(fmt)
+
+    logger = logging.getLogger()
+    logger.setLevel(logging.INFO)
+
+    # create stream handler
+    stream_handler = logging.StreamHandler(sys.stderr)
+    stream_handler.setFormatter(formatter)
+    stream_handler.setLevel(logging.INFO)
+
+    # create a file handler
+    file_handler = logging.FileHandler(utils.get_unique_filename(log_filename))
+    file_handler.setFormatter(formatter)
+
+    logger.addHandler(stream_handler)
+    logger.addHandler(file_handler)
+
+    for handler in handlers:
+        logger.addHandler(handler)

py_utils/models.py

@@ -0,0 +1,36 @@
+from datetime import datetime
+from enum import Enum
+from sqlmodel import Field, SQLModel
+from typing import Optional
+
+
+class JobStatusLevels(str, Enum):
+    INFO = "INFO"
+    DEBUG = "DEBUG"
+    ERROR = "ERROR"
+
+
+class JobStatus(SQLModel, table=True):
+    __tablename__: str = "job_status"
+
+    id: Optional[int] = Field(default=None, primary_key=True)
+    created_date: datetime = Field(default=datetime.utcnow(), nullable=False)
+    modified_date: datetime = Field(default_factory=datetime.utcnow, nullable=False)
+    host: str
+    script_path: str
+    script_name: str
+    executed_by: str
+    script_start_time: datetime
+    script_end_time: datetime
+    elapsed_time: int
+    job_summary_data: str
+    level: JobStatusLevels
+
+    def __str__(self):
+        return (
+            f"script_name: {self.script_name}, executed by: {self.executed_by}"
+            f" level: {self.level},"
+            f" job_summary_data: {self.job_summary_data},"
+            f" start_time: {self.script_start_time}, end_time: {self.script_end_time},"
+            f" elapsed_time: {self.elapsed_time}"
+        )

py_utils/utils.py

@@ -1,9 +1,17 @@
+from datetime import datetime
 from email.message import EmailMessage
 from email.policy import SMTP
+from .models import JobStatus, JobStatusLevels
+from .orm import DbClient
+from typing import Dict
 
+import getpass
+import json
+import pytz
 import os
 import re
 import smtplib
+import socket
 
 
 def _contains_html(content: str) -> bool:
@@ -128,3 +136,68 @@ def send_email(
     else:
         with smtplib.SMTP(host) as server:
             server.send_message(msg)
+
+
+class ScriptHelper():
+    """A helper class to log `JobStatus`.
+
+    An instance of `ScriptHelper` should be instantiated at the beginning of a script i.e., `helper = ScriptHelper()`,
+    and then used to call `log_failed_job` or `log_successful_job` when the script fails or completes. Internally,
+    `ScriptHelper` automatically captures `start_time`, `end_time`, `executed_by`, and `elapsed_time`, `script_path`,
+    `host`, and provides that information when attempting to log the `JobStatus`
+
+    References:
+    - The structure of `JobStatus` can be seen in `py_custodian/models.py`
+    """
+
+    def __init__(self, script_name: str, db_client: DbClient, tz: pytz.tzinfo.BaseTzInfo = pytz.utc):
+        """Creates an instance of ScriptHelper defaulting the timezone to use UTC.
+
+        Args:
+            script_name (str): The name of the script that is creating an instance of this class.
+            db_client (DbClient): The database client to write logs with.
+            tz (pytz.tzinfo.BaseTzInfo, optional): The timezone to use for datetime fields. Defaults to pytz.utc.
+        """
+        self.__tz = tz
+        self.__start_time = datetime.now(self.__tz)
+        self.__parent_script = script_name
+        self.__executed_by = getpass.getuser()
+
+        self._db_client = db_client
+        self._db_client.create_tables()
+
+    def log_failed_job(self, summary_data: dict, error: str):
+        """Attempt to log a failed `JobStatus` entry to the log database.
+
+        Args:
+            summary_data (Dict): Summary data to capture in the log entry.
+        """
+        job_status = self._get_job_status_of_type(summary_data, error, False)
+        self._db_client.insert_data(job_status)
+
+    def log_successful_job(self, summary_data: dict):
+        """Attempt to log a successful `JobStatus` entry to the log database.
+
+        Args:
+            summary_data (Dict): Summary data to capture in the log entry.
+        """
+        job_status = self._get_job_status_of_type(summary_data, "", True)
+        self._db_client.insert_data(job_status)
+
+    def _get_job_status_of_type(self, summary_data: Dict, error: str, succeeded: bool):
+        level = JobStatusLevels.INFO if succeeded else JobStatusLevels.ERROR
+        end_time = datetime.now(self.__tz)
+        elapsed_time = int((end_time - self.__start_time).total_seconds())
+
+        job_status = JobStatus(
+            executed_by=self.__executed_by,
+            host=socket.gethostname(),
+            script_path=os.path.abspath(self.__parent_script),
+            script_name=self.__parent_script,
+            script_start_time=self.__start_time,
+            script_end_time=end_time,
+            elapsed_time=elapsed_time,
+            job_summary_data=json.dumps({"data": json.dumps(summary_data), "error": error}),
+            level=level,
+        )
+        return job_status

pyproject.toml

@@ -12,6 +12,7 @@ python = "^3.11"
 sqlalchemy = "^2.0.20"
 sqlmodel = "^0.0.12"
 deprecated = "^1.2.14"
+pytz = "^2024.1"
 
 [tool.poetry.urls]
 "Homepage" = "https://github.com/ctsit/PyUtils"