0
\$\begingroup\$

I'm using the below Python script to port few table data from a remote server (MySQL database) to my data warehouse server (PostgreSQL DB), The script is working fine (I'm using foreign data wrapper), so I thought of asking your help to make it in a more generalised way and if there is any better option for doing it rather than this way.

In this I have to declare the table structure; perhaps I missed a way to do it dynamically without declaring each table structure specifically?

And any other modifications?

import psycopg2
import os
import time
import MySQLdb
import sys
from pprint import pprint
from datetime import datetime
from psycopg2 import sql
from utils.config import Configuration as Config
from utils.postgres_helper import get_connection
from utils.utils import get_global_config


def psql_Source_fetch_and_DestInsert(cnx_msql,cnx_psql,msql, psql, msql_command, psql_command):
    print("Function Call..")
    msql.execute(msql_command)

    for row in msql:
        try:
            print("Insertion of rows..")
            print(row)
            psql.execute(psql_command, row)
        except psycopg2.Error as e:
            print ("Cannot execute the query!!", e.pgerror)
            sys.exit("Problem occured with the query!!!")
    cnx_psql.commit()


def dB_Fetch():

# MySQLdb connection
 try:
    source_host = 'magento'
    conf = get_global_config()
    cnx_msql = MySQLdb.connect(host=conf.get(source_host, 'host'),
                               user=conf.get(source_host, 'user'),
                               passwd=conf.get(source_host, 'password'),
                               port=int(conf.get(source_host, 'port')),
                               db=conf.get(source_host, 'db'))
 except mysql.connector.Error as e:
   print ("MYSQL: Unable to connect!", e.msg)
   sys.exit(1)

# Postgresql connection
 try:
   cnx_psql = get_connection(get_global_config(), 'pg_dwh')
 except psycopg2.Error as e:
   print('PSQL: Unable to connect!\n{0}').format(e)
   sys.exit(1)

# Cursors initializations
 cur_msql = cnx_msql.cursor()
 cur_psql = cnx_psql.cursor()

 try:
   print("creating table using cursor")

   SQL_create_Staging_schema="""CREATE SCHEMA IF NOT EXISTS staging AUTHORIZATION postgres;"""

   SQL_create_sales_flat_quote="""DROP TABLE IF EXISTS staging.sales_flat_quote;CREATE TABLE IF NOT EXISTS staging.sales_flat_quote
      (
          customer_id         BIGINT
        , entity_id           BIGINT
        , store_id            BIGINT
        , customer_email      TEXT
        , customer_firstname  TEXT
        , customer_middlename TEXT
        , customer_lastname   TEXT
        , customer_is_guest   BIGINT
        , customer_group_id   BIGINT
        , created_at          TIMESTAMP WITHOUT TIME ZONE
        , updated_at          TIMESTAMP WITHOUT TIME ZONE
        , is_active           BIGINT
        , items_count         BIGINT
        , items_qty           BIGINT
        , base_currency_code  TEXT
        , grand_total         NUMERIC(12,4)
        , base_to_global_rate NUMERIC(12,4)
        , base_subtotal       NUMERIC(12,4)
        , base_subtotal_with_discount   NUMERIC(12,4)
        )
   ;"""

   SQL_create_sales_flat_quote_item="""DROP TABLE IF EXISTS staging.sales_flat_quote_item;CREATE TABLE IF NOT EXISTS staging.sales_flat_quote_item
                 (store_id        INTEGER
        , row_total       NUMERIC
        , updated_at      TIMESTAMP WITHOUT TIME ZONE
        , qty             NUMERIC
        , sku             CHARACTER VARYING
        , free_shipping   INTEGER
        , quote_id        INTEGER
        , price           NUMERIC
        , no_discount     INTEGER
        , item_id         INTEGER
        , product_type    CHARACTER VARYING
        , base_tax_amount NUMERIC
        , product_id      INTEGER
        , name            CHARACTER VARYING
        , created_at      TIMESTAMP WITHOUT TIME ZONE
        ,base_row_total   NUMERIC
        ,base_discount_amount NUMERIC
        ,applied_rule_ids CHARACTER VARYING
        );"""

   SQL_create_catalog_product_flat_1="""DROP TABLE IF EXISTS staging.catalog_product_flat_1;CREATE TABLE IF NOT EXISTS staging.catalog_product_flat_1
                 (name              CHARACTER VARYING
        , sku               CHARACTER VARYING
        , type_id           CHARACTER VARYING
        , created_at        TIMESTAMP WITHOUT TIME ZONE
        , url_path          CHARACTER VARYING
        , price             NUMERIC
        , short_description CHARACTER VARYING
        , url_key           CHARACTER VARYING
        , thumbnail_label   CHARACTER VARYING
        , small_image       CHARACTER VARYING
        , thumbnail         CHARACTER VARYING

        );"""

   SQL_create_cataloginventory_stock_item="""DROP TABLE IF EXISTS staging.cataloginventory_stock_item;CREATE TABLE staging.cataloginventory_stock_item
     (
      qty numeric,
      max_sale_qty numeric,
      stock_status_changed_auto bigint,
      notify_stock_qty numeric,
      backorders bigint,
      is_in_stock bigint,
      item_id bigint NOT NULL,
      low_stock_date timestamp without time zone,
      min_qty numeric,
      manage_stock bigint,
      min_sale_qty numeric,
      product_id bigint
     );"""


   print("Creating  Schema...")   
   cur_psql.execute(SQL_create_Staging_schema)

   print("Creating tables...")
   cur_psql.execute(SQL_create_sales_flat_quote)

   cur_psql.execute(SQL_create_sales_flat_quote_item)
   cur_psql.execute(SQL_create_cataloginventory_stock_item)
   cnx_psql.commit();

   print("Fetching data from source server")
   #select from Magento & insert into DWH: SELECT & INSERT SQL statements are created as PAIRS for looping.
   commands = [
               ("""SELECT qty,max_sale_qty,stock_status_changed_auto,notify_stock_qty,backorders,is_in_stock,
            item_id,low_stock_date,min_qty,manage_stock,min_sale_qty,product_id from cataloginventory_stock_item;""",
              """INSERT INTO staging.cataloginventory_stock_item (qty,max_sale_qty,stock_status_changed_auto,notify_stock_qty,backorders,is_in_stock,
            item_id,low_stock_date,min_qty,manage_stock,min_sale_qty,product_id) 
              SELECT %s, %s, %s,%s,%s,%s,%s,%s,%s, %s, %s, %s;"""), 

              ("""SELECT customer_id, entity_id, store_id, customer_email , customer_firstname, customer_middlename
                 , customer_lastname , customer_is_guest, customer_group_id, created_at, updated_at, is_active
                 , items_count, items_qty, base_currency_code, grand_total, base_to_global_rate, base_subtotal
                 , base_subtotal_with_discount from sales_flat_quote
                 where is_active=1
                 AND items_count != '0'
                 AND DATE_FORMAT(updated_at, '%Y-%m-%d') >= CURDATE();""",
              """INSERT INTO staging.sales_flat_quote (customer_id, entity_id, store_id, customer_email , customer_firstname
                , customer_middlename, customer_lastname , customer_is_guest, customer_group_id, created_at, updated_at
                , is_active, items_count, items_qty, base_currency_code, grand_total, base_to_global_rate, base_subtotal
                , base_subtotal_with_discount) 
              SELECT %s, %s, %s,%s,%s,%s,%s,%s,%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s;"""),

            ("""SELECT store_id,row_total,updated_at,qty,sku,free_shipping,quote_id,price,no_discount,item_id,product_type
                       ,base_tax_amount,product_id,name,created_at,base_row_total,base_discount_amount,applied_rule_ids from sales_flat_quote_item WHERE DATE_FORMAT(updated_at, '%Y-%m-%d') >= CURDATE();""",
             """INSERT INTO staging.sales_flat_quote_item (store_id,row_total,updated_at,qty,sku,free_shipping,quote_id
                                                          ,price,no_discount ,item_id,product_type,base_tax_amount,product_id,name
                                                          ,created_at,base_row_total,base_discount_amount,applied_rule_ids)
                SELECT %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s;""")

             ,("""SELECT created_at,url_path,price,short_description,url_key,thumbnail_label,small_image,thumbnail
                        ,name,sku,type_id from catalog_product_flat_1;""", 
             """INSERT INTO staging.catalog_product_flat_1 (created_at,url_path,price,short_description,url_key,thumbnail_label
                                                            ,small_image,thumbnail,name,sku,type_id) 
                SELECT %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s;""")            
            ]

   for msql_command, psql_command in commands:
       psql_Source_fetch_and_DestInsert(cnx_msql,cnx_psql,cur_msql, cur_psql, msql_command, psql_command)

 except (Exception, psycopg2.Error) as error:
     print ("Error while fetching data from PostgreSQL", error)
 finally:
     ## Closing cursors
     cur_msql.close()
     cur_psql.close()
     ## Committing
     cnx_psql.commit()
     ## Closing database connections
     cnx_msql.close()
     cnx_psql.close()


if __name__ == '__main__':
    dB_Fetch()
\$\endgroup\$
  • \$\begingroup\$ In my opinion, this question is going a bit beyond asking for a review of the code, in asking how to obtain the schema. That is more of a feature request. \$\endgroup\$ – 200_success Jul 24 at 12:19
  • \$\begingroup\$ @200_successGetting schema is fine, Can you please review this code and let me know what all changes that need to be made on this to make it better? \$\endgroup\$ – Linu Jul 25 at 7:49

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.