init

.gitignore

+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# pycharm
+.idea/
+
+# logs
+logs/
\ No newline at end of file

Dockerfile

+FROM python:3.7-slim
+ADD . /opt
+WORKDIR /opt
+RUN pip install -r requirements.txt
+CMD python main.py

conf/configuration.yml

+#---------------Service Configurations----------------#
+SERVICE_CONFIG:
+  LOG_LEVEL: info
+  LOG_HANDLER_NAME: PushDatatoPostgreSql
+  LOGSTASH_HOST: 192.168.1.47
+  LOGSTASH_PORT: 5000
+
+#--------------System Configurations--------------------#
+SYSTEM_CONFIG:
+  shared_mount_base_ai_job: /mnt/ilens/ai-jobs/
+  default_run_id: default_run_id
+

conf/dev-configuration.yml

+#---------------Service Configurations----------------#
+SERVICE_CONFIG:
+  LOG_LEVEL: info
+  LOG_HANDLER_NAME: PushDatatoPostgreSql
+  LOGSTASH_HOST: 192.168.1.47
+  LOGSTASH_PORT: 5000
+
+#--------------System Configurations--------------------#
+SYSTEM_CONFIG:
+  shared_mount_base_ai_job: /mnt/ilens/ai-jobs/
+  default_run_id: default_run_id
+
+#-----------Postgres Sql------------------------------#
+
+POSTGRES_CONFIG:
+  database: postgres
+  host: localhost
+  port: 5432
+

main.py

+import os
+import traceback
+from scripts.common.config_parser import config
+from scripts.common.logsetup import logger
+import psycopg2
+
+class PushDatatoPostgreSql:
+    def __init__(self, query):
+        self.query = query
+        self.conn = psycopg2.connect(
+            database=self.query['database'],
+            user='postgres',
+            password=password,
+            host='localhost',
+            port='5432'
+        )
+        logger.info("Connected to Postgres Sql....")
+        self.cur = self.conn.cursor()
+
+    def create_table(self, table_name):
+        try:
+            logger.info("Creating table with name " + table_name)
+            return True
+        except Exception as e:
+            logger.info(e)
+            logger.info("Table {} is already created............".format(table_name))
+            return False
+
+    def push_data(self):
+
+    def run(self):
+        try:
+            logger.info("Starting to push the data to Postgres...............")
+            # create table
+            
+            return True
+        except Exception as e:
+            logger.info(e)
+            logger.info(traceback.format_exc())
+            return False
+
+
+if __name__ == '__main__':
+    try:
+        obj = PushDatatoPostgreSql(config)
+        val = obj.run()
+        if val:
+            if len(os.listdir(config['component_output_dir'])) > 0:
+                logger.info("File created Successfully")
+            else:
+                logger.info("The output directory is empty")
+        else:
+            logger.info("Random Forest Component Failed")
+            logger.info(traceback.format_exc())
+    except Exception as e:
+        logger.info("Random Forest Component Failed")
+        logger.info(traceback.format_exc())

requirements.txt

+requests
+pyyaml
+python-logstash-async
+pymongo
+azure-storage-blob
\ No newline at end of file

scripts/common/config_parser.py

+#!/usr/bin/env python
+import os
+import sys
+import yaml
+import json
+from scripts.common.constants import PushDatatoPostgreSqlConstants
+
+config_path = os.path.join(os.getcwd(), "conf", "dev-configuration.yml")
+if os.path.exists(config_path):
+    sys.stderr.write("Reading config from --> {}".format(config_path))
+    sys.stderr.write("\n")
+
+    with open(config_path, 'r') as stream:
+        _config = yaml.safe_load(stream)
+else:
+    sys.stderr.write("Configuration not found...")
+    sys.stderr.write("Exiting....")
+    sys.exit(1)
+# uncomment for Testing
+# os.environ['pipeline_id'] = "pipe1"
+# os.environ['artifact_base_path'] = "/data/model/tested/"
+# os.environ['artifact_name'] = 'response.json'
+# ------------------------ Configurations -----------------------------------------------------------------------------
+pipeline_id = os.environ.get('PIPELINE_ID', default="pipeline_313")
+shared_mount_base_ai_job = os.environ.get("shared_mount_base_ai_job",
+                                          _config.get("SYSTEM_CONFIG", {}).get('shared_mount_base_ai_job'))
+
+# read from $shared_mount_base_ai_job/$pipeline_id/run.config
+run_id_path = os.path.join(shared_mount_base_ai_job, pipeline_id, "run_config.json")
+try:
+    sys.stderr.write("Checking for run id parameters at path " + run_id_path + "\n")
+    with open(run_id_path) as f:
+        run_id_param = json.load(f)
+    run_id = run_id_param['run_id']
+except Exception as e:
+    sys.stderr.write("No run_config.json file is there so keeping run id as default_run_id  " + "\n")
+    run_id = _config.get("SYSTEM_CONFIG", {}).get('default_run_id')
+
+historic_data_path = os.path.join(shared_mount_base_ai_job, 'historical_weather_data')
+
+# ----------------------Postgres----------------------------------------------------------------------------------------
+
+postgres_host = _config.get("POSTGRES_CONFIG", {}).get('host')
+postgres_port = int(_config.get("POSTGRES_CONFIG", {}).get('port'))
+postgres_database = _config.get("POSTGRES_CONFIG", {}).get('database')
+
+# ----------------------------------------------------------------------------------------------------------------------
+
+BASE_LOG_PATH = os.path.join(os.getcwd(), "logs")
+if not os.path.exists(os.path.join(os.getcwd(), 'logs')):
+    os.mkdir(os.path.join(os.getcwd(), 'logs'))
+
+LOG_LEVEL = os.environ.get("LOG_LEVEL", _config.get('SERVICE_CONFIG', {}).get("LOG_LEVEL", "INFO")).upper()
+LOG_HANDLER_NAME = _config.get('SERVICE_CONFIG', {}).get("LOG_HANDLER_NAME", "PushDatatoPostgreSql")
+ENABLE_LOGSTASH_LOG = os.environ.get("ENABLE_LOGSTASH_LOG", 'False').lower()
+LOGSTASH_HOST = _config.get('SERVICE_CONFIG', {}).get('LOGSTASH_HOST')
+LOGSTASH_PORT = str(_config.get('SERVICE_CONFIG', {}).get('LOGSTASH_PORT'))
+
+config = {
+    'pipeline_id': pipeline_id,
+    'run_id': run_id,
+    'shared_mount_base_ai_job': shared_mount_base_ai_job,
+    'component_output_dir': historic_data_path,
+}
+
+if not os.path.exists(config['shared_mount_base_ai_job']):
+    sys.stderr.write("Shared path does not exist!" + "\n")
+    sys.stderr.write("Creating path --> {}".format(config['shared_mount_base_ai_job'] + "\n"))
+    os.makedirs(config['shared_mount_base_ai_job'])
+
+if not os.path.exists(config['component_output_dir']):
+    sys.stderr.write("component_output_dir does not exist!" + "\n")
+    sys.stderr.write("Creating path --> {}".format(config['component_output_dir'] + "\n"))
+    os.makedirs(config['component_output_dir'])

scripts/common/constants.py

+#!/usr/bin/env python
+
+
+class PushDatatoPostgreSqlConstants:
+    NEXT_COMPONENT = ""
+    CONTAINER = "ilensqa"
+    CONNECTION_STRING = "DefaultEndpointsProtocol=https;AccountName=azrabsilensqa01;AccountKey=DN6q6kX98JM8yUwtuJh2bAaXUGFo1zRS5HJSsa/ZA+MlmctjC000eHP7bdXiQqkI/MVtADhS8c9E88LI5T4UHw==;EndpointSuffix=core.windows.net"
+    COMPONENT_NAME = "PushDatatoPostgreSql"
+    HTTP = "http://"
+    LOG_VAR_MESSAGE = "\n" + "#" * 25 + "\n" + "{}" + "\n" + "#" * 25 + "\n" + "{}\n"
+
+
+class ComponentExceptions:
+    INVALID_AZURE_FILE_PATH_EXCEPTION = "AZURE PATH ERROR"
+    INVALID_LOCAL_FILE_PATH_EXCEPTION = "No File in the local path"
+    INVALID_ARTIFACT_BASE_PATH_EXCEPTION = "Artifact base path value is missing"
+    INVALID_AZURE_FILE_NAME_EXCEPTION = "Artifact name is missing"
+    INVALID_CONTAINER_NAME = "Container name is missing"

scripts/common/logsetup.py

+import os
+import logging
+from logging.handlers import RotatingFileHandler
+from logstash_async.handler import AsynchronousLogstashHandler
+from scripts.common.config_parser import LOG_LEVEL, LOG_HANDLER_NAME, BASE_LOG_PATH
+from scripts.common.config_parser import LOG_LEVEL, LOG_HANDLER_NAME, BASE_LOG_PATH, LOGSTASH_HOST, LOGSTASH_PORT, ENABLE_LOGSTASH_LOG
+
+DEFAULT_FORMAT = '%(asctime)s %(levelname)5s %(name)s %(message)s'
+DEBUG_FORMAT = '%(asctime)s %(levelname)5s %(name)s [%(threadName)5s:%(filename)5s:%(funcName)5s():%(lineno)s] %(' \
+               'message)s '
+EXTRA = {}
+FORMATTER = DEFAULT_FORMAT
+if LOG_LEVEL.strip() == "DEBUG":
+    FORMATTER = DEBUG_FORMAT
+logging.trace = logging.DEBUG - 5
+logging.addLevelName(logging.DEBUG - 5, 'TRACE')
+
+
+class ILensLogger(logging.getLoggerClass()):
+    def __init__(self, name):
+        super().__init__(name)
+
+    def trace(self, msg, *args, **kwargs):
+        if self.isEnabledFor(logging.trace):
+            self._log(logging.trace, msg, args, **kwargs)
+
+
+def get_logger(log_handler_name):
+    """
+    Purpose : To create logger .
+    :param log_handler_name: Name of the log handler.
+    :return: logger object.
+    """
+    log_path = os.path.join(BASE_LOG_PATH, log_handler_name + ".log")
+    logging.setLoggerClass(ILensLogger)
+    _logger = logging.getLogger(log_handler_name)
+    _logger.setLevel(LOG_LEVEL.strip().upper())
+    log_handler = logging.StreamHandler()
+    log_handler.setLevel(LOG_LEVEL)
+    formatter = logging.Formatter(FORMATTER)
+    log_handler.setFormatter(formatter)
+    handler = RotatingFileHandler(log_path, maxBytes=10485760,
+                                  backupCount=5)
+    handler.setFormatter(formatter)
+    _logger.addHandler(log_handler)
+    _logger.addHandler(handler)
+    if ENABLE_LOGSTASH_LOG == 'true' and LOGSTASH_PORT is not None and LOGSTASH_HOST is not None and LOGSTASH_PORT.isdigit():
+        _logger.addHandler(AsynchronousLogstashHandler(LOGSTASH_HOST, int(LOGSTASH_PORT), database_path=None))
+
+    return _logger
+
+
+logger = get_logger(LOG_HANDLER_NAME)