db_creation_module.py

# Personal License Agreement
# Copyright Notice
#
# © 2026 Voltaire Bazurto Blacio. All rights reserved.
# License Terms
#
#     Ownership: All code contained in this portfolio is the sole property of Voltaire Bazurto Blacio and is hereby copyrighted by me.
#
#     Permitted Use: Others are welcome to view and study the code for personal, educational, or non-commercial purposes. You may share insights or information about the code, but you cannot use it for any commercial products, either as-is or in a derivative form.
#
#     Restrictions: The code may not be used, reproduced, or distributed for commercial purposes under any circumstances without my explicit written permission.
#
#     Rights Reserved: I reserve the right to use the code, or any future versions thereof, for my own purposes in any way I choose, including but not limited to the development of future commercial derivative works under my name or personal brand.
#
#     Disclaimer: The code is provided "as is" without warranty of any kind, either express or implied. I am not responsible for any damages resulting from the use of this code.
#
# By accessing this portfolio, you agree to abide by these terms.

import os.path
from botocore.exceptions import ClientError
import wget
class HRRecruitAppDBCreator:
    __tmp_sql_dir = './tmp_sql_dir'
    __sql_base_url = 'https://raw.githubusercontent.com/vbazurtob/HRRecruitApp/refs/heads/master/sql_data/'
    __sql_db_structure_files = (
        '0000_database.sql',
        '0001_schema.sql',
        '0002_tables.sql',


    )

    __sql_db_data_files = (
        '0003_initialData.sql',
        '0004_testData.sql',
    )

    __sql_db_security_files = (
        '0005_initial_app_user.sql',
        '0006_security.sql'
    )

    def __init__(self, db_instance_id):
        self.db_instance_id = db_instance_id

    def download_sql_files(self, sql_filename):
        if not os.path.exists(self.__tmp_sql_dir):
            os.mkdir(self.__tmp_sql_dir)
        output_file = os.path.join(self.__tmp_sql_dir, sql_filename)
        sql_content = wget.download(f'{self.__sql_base_url}/{sql_filename}', out=output_file)
        return sql_content

    def execute_single_sql_file(self, db_connection, sql_content):
        db_connection.cursor().execute(sql_content)
        db_connection.commit()

    def execute_sql_files(self, db_connection):
        # clean old files if the tmp dir exists
        if os.path.exists(self.__tmp_sql_dir):
            for sql_filename in os.listdir(self.__tmp_sql_dir):
              os.unlink(os.path.join(self.__tmp_sql_dir, sql_filename))

        #1. Structure
        for filename in self.__sql_db_structure_files:
            self.download_sql_files(filename)
            downloaded_file = os.path.join(self.__tmp_sql_dir, filename)
            if not os.path.exists( downloaded_file ):
                raise
            else:
                try:
                  opened_file = open(downloaded_file, 'r')
                  sql_file_content = opened_file.read()
                except Exception as e:
                    print(f'Error trying to read file {filename}\n{e}\n')
                    exit(1)
                try:
                    self.execute_single_sql_file(db_connection, sql_file_content)
                    print(f'Executed file {filename}')
                except Exception as sql_e:
                    print(f'Error executing SQL file {filename}\n{sql_e}\n')
                    exit(1)
                if opened_file:
                    opened_file.close()

        db_connection.close()

    def get_secret(self, client, secret_name):
        try:
            get_secret_value_response = client.get_secret_value(
                SecretId=secret_name
            )
        except ClientError as e:
            raise e

        secret = get_secret_value_response['SecretString']
        return secret