diff --git a/ush/python/pygw/src/pygw/logger.py b/ush/python/pygw/src/pygw/logger.py
new file mode 100644
index 0000000000..5807d5cd94
--- /dev/null
+++ b/ush/python/pygw/src/pygw/logger.py
@@ -0,0 +1,223 @@
+"""
+Logger
+"""
+
+import sys
+from pathlib import Path
+from typing import Union, List
+import logging
+
+
+class ColoredFormatter(logging.Formatter):
+    """
+    Logging colored formatter
+    adapted from https://stackoverflow.com/a/56944256/3638629
+    """
+
+    grey = '\x1b[38;21m'
+    blue = '\x1b[38;5;39m'
+    yellow = '\x1b[38;5;226m'
+    red = '\x1b[38;5;196m'
+    bold_red = '\x1b[31;1m'
+    reset = '\x1b[0m'
+
+    def __init__(self, fmt):
+        super().__init__()
+        self.fmt = fmt
+        self.formats = {
+            logging.DEBUG: self.blue + self.fmt + self.reset,
+            logging.INFO: self.grey + self.fmt + self.reset,
+            logging.WARNING: self.yellow + self.fmt + self.reset,
+            logging.ERROR: self.red + self.fmt + self.reset,
+            logging.CRITICAL: self.bold_red + self.fmt + self.reset
+        }
+
+    def format(self, record):
+        log_fmt = self.formats.get(record.levelno)
+        formatter = logging.Formatter(log_fmt)
+        return formatter.format(record)
+
+
+class Logger:
+    """
+    Improved logging
+    """
+    LOG_LEVELS = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
+    DEFAULT_LEVEL = 'INFO'
+    DEFAULT_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-12s: %(message)s'
+    def __init__(self, name: str = None,
+                 level: str = DEFAULT_LEVEL,
+                 _format: str = DEFAULT_FORMAT,
+                 colored_log: bool = False,
+                 logfile_path: Union[str, Path] = None):
+        """
+        Initialize Logger
+
+        Parameters
+        ----------
+        name         : str
+                       Name of the Logger object
+                       default : None
+        level        : str
+                       Desired Logging level
+                       default : 'INFO'
+        _format      : str
+                       Desired Logging Format
+                       default : '%(asctime)s - %(levelname)-8s - %(name)-12s: %(message)s'
+        colored_log  : bool
+                       Use colored logging for stdout
+                       default: False
+        logfile_path : str or Path
+                       Path for logging to a file
+                       default : None
+        """
+
+        self.name = name
+        self.level = level.upper()
+        self.format = _format
+        self.colored_log = colored_log
+
+        if self.level not in Logger.LOG_LEVELS:
+            raise LookupError('{self.level} is unknown logging level\n' +
+                              'Currently supported log levels are:\n' +
+                              f'{" | ".join(Logger.LOG_LEVELS)}')
+
+        # Initialize the root logger if no name is present
+        self._logger = logging.getLogger(name) if name else logging.getLogger()
+
+        self._logger.setLevel(self.level)
+
+        _handlers = []
+        # Add console handler for logger
+        _handler = Logger.add_stream_handler(
+            level=self.level,
+            _format=self.format,
+            colored_log=self.colored_log,
+            )
+        _handlers.append(_handler)
+        self._logger.addHandler(_handler)
+
+        # Add file handler for logger
+        if logfile_path is not None:
+            _handler = Logger.add_file_handler(logfile_path, level=self.level, _format=self.format)
+            self._logger.addHandler(_handler)
+            _handlers.append(_handler)
+
+    def __getattr__(self, attribute):
+        """
+        Allows calling logging module methods directly
+
+        Parameters
+        ----------
+        attribute : str
+                    attribute name of a logging object
+
+        Returns
+        -------
+        attribute : logging attribute
+        """
+        return getattr(self._logger, attribute)
+
+    def get_logger(self):
+        """
+        Return the logging object
+
+        Returns
+        -------
+        logger : Logger object
+        """
+        return self._logger
+
+    @classmethod
+    def add_handlers(cls, logger: logging.Logger, handlers: List[logging.Handler]):
+        """
+        Add a list of handlers to a logger
+
+        Parameters
+        ----------
+        logger : logging.Logger
+                 Logger object to add a new handler to
+        handlers: list
+                 A list of handlers to be added to the logger object
+
+        Returns
+        -------
+        logger : Logger object
+        """
+        for handler in handlers:
+            logger.addHandler(handler)
+
+        return logger
+
+    @classmethod
+    def add_stream_handler(cls, level: str = DEFAULT_LEVEL,
+                            _format: str = DEFAULT_FORMAT,
+                            colored_log: bool = False):
+        """
+        Create stream handler
+        This classmethod will allow setting a custom stream handler on children
+
+        Parameters
+        ----------
+        level : str
+                logging level
+                default : 'INFO'
+        _format : str
+                  logging format
+                  default : '%(asctime)s - %(levelname)-8s - %(name)-12s: %(message)s'
+        colored_log : bool
+                      enable colored output for stdout
+                      default : False
+
+        Returns
+        -------
+        handler : logging.Handler
+                  stream handler of a logging object
+        """
+
+        handler = logging.StreamHandler(sys.stdout)
+        handler.setLevel(level)
+        _format = ColoredFormatter(_format) if colored_log else logging.Formatter(_format)
+        handler.setFormatter(_format)
+
+        return handler
+
+    @classmethod
+    def add_file_handler(cls, logfile_path: Union[str, Path],
+                         level: str = DEFAULT_LEVEL,
+                         _format: str = DEFAULT_FORMAT):
+        """
+        Create file handler.
+        This classmethod will allow setting custom file handler on children
+        Create stream handler
+        This classmethod will allow setting a custom stream handler on children
+
+        Parameters
+        ----------
+        logfile_path: str or Path
+                      Path for writing out logfiles from logging
+                      default : False
+        level : str
+                logging level
+                default : 'INFO'
+        _format : str
+                  logging format
+                  default : '%(asctime)s - %(levelname)-8s - %(name)-12s: %(message)s'
+
+        Returns
+        -------
+        handler : logging.Handler
+                  file handler of a logging object
+        """
+
+        logfile_path = Path(logfile_path)
+
+        # Create the directory containing the logfile_path
+        if not logfile_path.parent.is_dir():
+            logfile_path.mkdir(parents=True, exist_ok=True)
+
+        handler = logging.FileHandler(str(logfile_path))
+        handler.setLevel(level)
+        handler.setFormatter(logging.Formatter(_format))
+
+        return handler
diff --git a/ush/python/pygw/src/pygw/task.py b/ush/python/pygw/src/pygw/task.py
new file mode 100644
index 0000000000..14d365cbec
--- /dev/null
+++ b/ush/python/pygw/src/pygw/task.py
@@ -0,0 +1,60 @@
+class Task:
+    """
+    Base class for all tasks
+    """
+
+    def __init__(self, config, *args, **kwargs):
+        """
+        Every task needs a config.
+        Additional arguments (or key-value arguments) can be provided.
+
+        Parameters
+        ----------
+        config : Dict
+                 dictionary object containing task configuration
+
+        *args : tuple
+                Additional arguments to `Task`
+
+        **kwargs : dict, optional
+                   Extra keyword arguments to `Task`
+        """
+
+        # Store the config and arguments as attributes of the object
+        self.config = config
+
+        for arg in args:
+            setattr(self, str(arg), arg)
+
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+    def initialize(self):
+        """
+        Initialize methods for a task
+        """
+        pass
+
+    def configure(self):
+        """
+        Configuration methods for a task in preparation for execution
+        """
+        pass
+
+    def execute(self):
+        """
+        Execute methods for a task
+        """
+        pass
+
+    def finalize(self):
+        """
+        Methods for after the execution that produces output task
+        """
+        pass
+
+    def clean(self):
+        """
+        Methods to clean after execution and finalization prior to closing out a task
+        """
+        pass
diff --git a/ush/python/pygw/src/tests/test_logger.py b/ush/python/pygw/src/tests/test_logger.py
new file mode 100644
index 0000000000..2cce0f7449
--- /dev/null
+++ b/ush/python/pygw/src/tests/test_logger.py
@@ -0,0 +1,42 @@
+from pygw.logger import Logger
+
+level = 'debug'
+number_of_log_msgs = 5
+reference = {'debug': "Logging test has started",
+             'info': "Logging to 'logger.log' in the script dir",
+             'warning': "This is my last warning, take heed",
+             'error': "This is an error",
+             'critical': "He's dead, She's dead.  They are all dead!"}
+
+
+def test_logger(tmp_path):
+    """Test log file"""
+
+    logfile = tmp_path / "logger.log"
+
+    try:
+        log = Logger('test_logger', level=level, logfile_path=logfile, colored_log=True)
+        log.debug(reference['debug'])
+        log.info(reference['info'])
+        log.warning(reference['warning'])
+        log.error(reference['error'])
+        log.critical(reference['critical'])
+    except Exception as e:
+        raise AssertionError(f'logging failed as {e}')
+
+    # Make sure log to file created messages
+    try:
+        with open(logfile, 'r') as fh:
+            log_msgs = fh.readlines()
+    except Exception as e:
+        raise AssertionError(f'failed reading log file as {e}')
+
+    # Ensure number of messages are same
+    log_msgs_in_logfile = len(log_msgs)
+    assert log_msgs_in_logfile == number_of_log_msgs
+
+    # Ensure messages themselves are same
+    for count, line in enumerate(log_msgs):
+        lev = line.split('-')[3].strip().lower()
+        message = line.split(':')[-1].strip()
+        assert reference[lev] == message