# from flask import Flask import os import psycopg2 from dotenv import load_dotenv from flask_swagger_ui import get_swaggerui_blueprint import datetime import jwt import random import json import names from flask import ( Flask, jsonify, send_from_directory, request, ) from flask_cors import CORS, cross_origin from datetime import timedelta, date, datetime from functools import wraps import json import psycopg2.extras import pandas as pd from flask_sqlalchemy import SQLAlchemy from sqlalchemy import create_engine #from datetime import datetime # import socket #import SQLAlchemy #from werkzeug.utils import secure_filename load_dotenv() app = Flask(__name__) cors = CORS(app) # allow CORS for all domains on all routes. SWAGGER_URL = '/api/docs' # URL for exposing Swagger UI (without trailing '/') # API_URL = 'http://petstore.swagger.io/v2/swagger.json' # Our API url (can of course be a local resource) # API_URL = 'http://localhost:6335/docs/digifi_swagger.json' API_URL = 'https://chiefsoft.com/docs/simbrella/digifi_swagger.json' # Call factory function to create our blueprint swaggerui_blueprint = get_swaggerui_blueprint( SWAGGER_URL, # Swagger UI static files will be mapped to '{SWAGGER_URL}/dist/'m API_URL, config={ # Swagger UI config overrides 'app_name': "digiFi-Core" }, # oauth_config={ # OAuth config. See https://github.com/swagger-api/swagger-ui#oauth2-configuration . # 'clientId': "your-client-id", # 'clientSecret': "your-client-secret-if-required", # 'realm': "your-realms", # 'appName': "your-app-name", # 'scopeSeparator': " ", # 'additionalQueryStringParams': {'test': "hello"} # } ) app.register_blueprint(swaggerui_blueprint) dataUrl = os.getenv("DATABASE_URL") connection = psycopg2.connect(dataUrl) @app.route('/') def hello(): return "Hello World!" @app.route('/salary/login', methods=["POST"]) def salary_login(): try: data = request.json if not data: return { "message": "Please provide user details", "data": None, "error": "Bad request" }, 400 # validate input username = data["username"] password = data["password"] if username == 'digifiuser' and password == 'digifipass' : return { "message": "Successfully Login", "data": [ {"uid":"425611f2-c692-4404-b93d-76ca7a5ce70","icon": "icon_user", "name": "Simulator User" , "status": '1', "id": 1} ] }, 201 else: return { "message": "Error fetching auth token!, invalid email or password", "data": None, "error": "Unauthorized" }, 404 except Exception as e: return { "message": "Something went wrong!", "error": str(e), "data": None }, 500 @app.route('/salary/verifypin', methods=["POST"]) def salary_verifypin(): try: data = request.json if not data: return { "message": "Please provide pin", "data": None, "error": "Bad request" }, 400 # validate input bvn = data["bvn"] pin = data["pin"] SELECT_ACC = "SELECT id, uid,pin,bvn FROM demo_bank_accounts WHERE bvn='" + bvn + "' AND pin = '" + pin + "' " with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_ACC) account = cursor.fetchall() account_found = count = len( account ) if account_found == 1 : return { "status": "1", "data": account }, 201 else: return { "status": "-1", "data": None, "message" : "Invalid Pin", "error": "1" }, 404 except Exception as e: return { "status": "0", "error": str(e), "message" : "Something went wrong", "data": None }, 500 @app.route('/salary/demousers') def salary_demousers(): try: SELECT_DEMO_ENTRY = f"SELECT id,uid, bvn AS CustomerID, mobile AS AccountID, name,offers,salary_account,current_loans,mobile,bvn, email, pin, added::text, balance, updated::text AS updated FROM demo_bank_accounts ORDER BY id DESC LIMIT 1500" print(SELECT_DEMO_ENTRY) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_DEMO_ENTRY) select_demoS = cursor.fetchall() # "last_update": datetime.datetime.utcnow(),# demo_data = json.dumps([dict(ix) for ix in select_demoS]) result_demo_data = { "last_update": "2010-01-01", "offers": products(), "list": json.loads(demo_data) } return { "demo_data": result_demo_data, }, 200 except Exception as e: return { "demo_data": [], "error": f"ERROR: {e}" }, 200 # VARCHAR(125), # offers INT DEFAULT 0, # INT DEFAULT 0, # current_loans INT DEFAULT 0, # VARCHAR(25), # VARCHAR(12), # VARCHAR(125), # VARCHAR(6), # SQL_INSERT = "INSERT INTO demo_bank_accounts (name, salary_account, mobile, bvn, email, pin) VALUES(%s,%s,%s,%s,%s,%s)" # for i in range(900): # new_data = ( # names.get_full_name() , # random.randint(0, 1), # '801-000-'+str( random.randint(1000, 9999) ) , # '8315000'+str( random.randint(1000, 9999) ), # 'demo+'+ names.get_first_name() +'@chiefsoft.net', # random.randint(1100, 9999) # ) # # print(new_data) # with connection: # with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: # cursor.execute(SQL_INSERT, new_data) # # print(names.get_full_name()) # return jsonify(demo_data=demo_data) # print(demo_data) # sample_range = random.randint(2, 5) # for x in range(sample_range): # calDate = datetime.datetime.utcnow() + datetime.timedelta(minutes=180 * random.randint(1, 20)) # new_l = { # "uid":"425611f2-c692-4404-b93d-76ca7a5ce7"+str(x), # "name": names.get_full_name() , # "added": calDate, # "offers":random.randint(1, 4), # "mobile": '801-000-'+str( random.randint(1000, 9999) ) , # "bvn": '8315000'+str( random.randint(1000, 9999) ), # "email": 'demo+'+ names.get_first_name() +'@chiefsoft.net' , # "current_loans" : [], # "pin": random.randint(1100, 9999) , # "salary_account": random.randint(0, 1) # } # dList.append(new_l) @app.route('/salary/products') def salary_products(): product_data = { "products": products() , "extra" : [] } return { "product_data": product_data, }, 200 @app.route('/salary/loanoffers') def salary_loanoffers2(): offers_data = { "offers": offers() , "extra" : [] } return { "product_data": offers_data, }, 200 @app.route('/salary/loanselect', methods=["POST"]) def salary_loanselect2(): try: data = request.json if not data: return { "message": "Please provide bvn and loan", "data": None, "error": "Bad request" }, 400 # validate input loan = data["loan"] bvn = data["bvn"] print ("here - 001 ") if loan != '' and bvn != '' : print ("here - 002 ") loan_result = load_offer(loan,bvn) INSERT_LOAN ="INSERT INTO loan_select (loan, bvn) VALUES(%s, %s)" new_data = (loan,bvn) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(INSERT_LOAN, new_data) print ("here - 003 ") return { "message": "REQUEST_AMOUNT", "loan": loan, "active_loan_count": 0, "active_loan": [] }, 201 else: return { "message": "Error fetching loan!, invalid bvn or loan", "data": None, "error": "Unauthorized" }, 404 except Exception as e: return { "message": "Something went wrong!", "error": str(e), "data": None }, 500 @app.route('/salary/loanapply', methods=["POST"]) def salary_loanapply2(): try: data = request.json if not data: return { "message": "Please provide bvn and loan and amount", "data": None, "error": "Bad request" }, 400 # validate input loan = data["loan"] bvn = data["bvn"] amount = data["amount"] if loan != '' and bvn != '' and amount > 0 : loan_read = load_offer(loan,bvn) UPDATE_LOAN ="UPDATE loan_apply SET status=3 WHERE bvn = '" + bvn + "' AND status = 1 " INSERT_LOAN ="INSERT INTO loan_apply (loan, bvn, amount) VALUES(%s, %s, %s)" SELECT_LOAN ="SELECT uid FROM loan_apply WHERE loan = '" + loan + "' AND bvn = '" + bvn + "' AND status = 1 ORDER BY id DESC LIMIT 1" new_data = (loan,bvn, amount) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(INSERT_LOAN, new_data) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_LOAN) loan_record = cursor.fetchall() loan_found = count = len( loan_record ) return { "message": "REQUEST_PIN", "bvn" : bvn, "loan_application_id": loan_record[0], "active_loan_count": 0, "active_loan": [] }, 201 else: return { "message": "Error fetching loan!, invalid bvn or loan or amount", "data": None, "error": "Unauthorized" }, 404 except Exception as e: return { "message": "Something went wrong!", "error": str(e), "data": None }, 500 @app.route('/salary/verifloan', methods=["POST"]) def salary_verifloan2(): try: data = request.json if not data: return { "message": "Please provide pin, bvn and loan_application_id", "data": None, "error": "Bad request" }, 400 # validate input bvn = data["bvn"] pin = data["pin"] loan_application_id = data["loan_application_id"] # loan_read = load_offer(loan,bvn) SELECT_ACC = "SELECT id, uid,pin,bvn FROM demo_bank_accounts WHERE bvn='" + bvn + "' AND pin = '" + pin + "' " print(SELECT_ACC) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_ACC) account = cursor.fetchall() account_found = count = len( account ) print("10000-a") if account_found == 1 : UPDATE_APPLICATION = "UPDATE loan_apply SET verified = now(), due_date = now() + '30 days' , status=5 WHERE uid::text = '" + loan_application_id + "' AND status = 1" print(UPDATE_APPLICATION) with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(UPDATE_APPLICATION) print("10000-cccc") loan_schedule = loan_create_final(bvn,loan_application_id) return { "qr": UPDATE_APPLICATION, "status": "1", "message" : "Loan Verified", "data": account }, 201 else: return { "status": "-1", "data": None, "message" : "Invalid Loan or PIN", "error": "1" }, 404 # let us verify the loan too with the UID except Exception as e: return { "status": "0", "error": str(e), "message" : "Something went wrong", "data": None }, 500 def loan_create_final(bvn,loan_application_uid): SELECT_LOAN = "SELECT uid,loan,bvn,amount,added::text,verified::text,status,due_date::text FROM loan_apply WHERE uid::text = '" + loan_application_uid + "' AND bvn ='" + bvn + "' " print(SELECT_LOAN) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_LOAN) select_demoS = cursor.fetchall() loan_data = json.dumps( [dict(ix) for ix in select_demoS] ) loan_result = json.loads( loan_data ) print(loan_result) loan=loan_result[0]['loan'] approved_amount=loan_result[0]['amount'] days_duration=30 due_date= date.today() + timedelta(days=days_duration) INSERT_LOAN="INSERT INTO loans (application_uid,bvn,loan,approved_amount,days_duration,due_date) VALUES (%s,%s,%s,%s,%s,%s)" new_data = (loan_application_uid,bvn,loan,approved_amount,days_duration,due_date) #print(new_data) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(INSERT_LOAN,new_data) return loan_result def products(): product_data = [ {"cid": "1", "description": "Product Loan 01" , "active" : 0 }, {"cid": "2", "description": "Product Loan 02" , "active" : 0 }, {"cid": "3", "description": "First Advance" , "active" : 1 }, ] return product_data # CREATE TABLE loans ( # id SERIAL, # uid uuid DEFAULT uuid_generate_v4(), # application_uid VARCHAR(150) NOT NULL, # bvn VARCHAR(12) NOT NULL, # loan VARCHAR(25) REFERENCES loan_offers (loan), # approved_amount INT DEFAULT 0, # days_duration INT DEFAULT 0, # status INT DEFAULT 1, # due_date timestamp, # payment INT DEFAULT 0, # added timestamp without time zone DEFAULT now() # ); # ALTER TABLE ONLY loans # ADD CONSTRAINT loans_id_key UNIQUE (id); def load_offer(loan,bvn): OFFER_QUERY = "SELECT uid AS cid,loan,amount,description,days_duration,active FROM loan_offers WHERE loan='" + loan +"' LIMIT 1" with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(OFFER_QUERY) select_demoS = cursor.fetchall() loan_data = json.dumps( [dict(ix) for ix in select_demoS] ) loan_result = json.loads( loan_data ) return json.loads( loan_data ) def offers(): OFFER_QUERY = "SELECT uid AS cid,loan,amount,description,days_duration,active FROM loan_offers ORDER BY amount ASC" with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(OFFER_QUERY) select_demoS = cursor.fetchall() demo_data = json.dumps( [dict(ix) for ix in select_demoS] ) return json.loads( demo_data) # # offers_data = [ # {"cid": "425611f2-c692-4404-b93d-76ca7a5ce00", "description": "100,000 Naira for 30 Days" , "active" : 1 }, # {"cid": "425611f2-c692-4404-b93d-76ca7a5ce01", "description": "300,000 Naira for 60 Days" , "active" : 1 }, # {"cid": "425611f2-c692-4404-b93d-76ca7a5ce02", "description": "900,000 Naira for 90 Days" , "active" : 1 }, # ] # return offers_data def offers_detail(offer_id): offer_detail = [ {"cid": "425611f2-c692-4404-b93d-76ca7a5ce00", "description": "100,000 Naira for 30 Days" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce01", "description": "300,000 Naira for 60 Days" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce02", "description": "900,000 Naira for 90 Days" , "active" : 1 }, ] return offer_detail @app.route('/office/auth') def office_login(): result_data = { "data": [], "extra" : [] } return { "result_data": result_data, }, 200 @app.route('/office/loan/select') def loan_select(): result_data = { "data": office_loan_data_select('SELECT'), "extra" : [] } return { "result_data": result_data, }, 200 @app.route('/office/loan/apply') def loan_apply(): result_data = { "data": office_loan_data('APPLY'), "extra" : [] } return { "result_data": result_data, }, 200 @app.route('/office/loan/approved') def loan_approved(): result_data = { "data": office_loan_data('APPROVED'), "extra" : [] } return { "result_data": result_data, }, 200 @app.route('/office/loan/disbursements') def loan_disbursements(): result_data = { "data": office_loan_data('DISBURSEMENTS'), "extra" : [] } return { "result_data": result_data, }, 200 #Disbursements def office_loan_data_select(loanLevel): SELECT_Q = f'''SELECT ls.id,ls.added::text, dm.name,dm.mobile, ls.bvn , ls.loan,lo.description FROM loan_select ls LEFT JOIN loan_offers lo ON lo.loan=ls.loan LEFT JOIN demo_bank_accounts dm ON dm.bvn = ls.bvn ORDER BY ls.id DESC LIMIT 300 ''' with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_Q) select_demoS = cursor.fetchall() demo_data = json.dumps( [dict(ix) for ix in select_demoS] ) return json.loads( demo_data) def office_loan_data(loanLevel): # public | loan_apply | table | salaryloan # public | loan_select | table | salaryloan SELECT_Q = f'''SELECT ls.id,ls.added::text, dm.name,dm.mobile, ls.bvn , ls.loan,lo.description FROM loan_select ls LEFT JOIN loan_offers lo ON lo.loan=ls.loan LEFT JOIN demo_bank_accounts dm ON dm.bvn = ls.bvn ORDER BY ls.id DESC LIMIT 300 ''' if loanLevel == "APPLY": SELECT_Q = f'''SELECT ls.id,ls.added::text, dm.name,dm.mobile, ls.bvn , ls.loan,lo.description,ls.amount,ls.verified::text FROM loan_apply ls LEFT JOIN loan_offers lo ON lo.loan=ls.loan LEFT JOIN demo_bank_accounts dm ON dm.bvn = ls.bvn WHERE ls.status = 1 ORDER BY ls.id DESC LIMIT 300''' if loanLevel == "APPROVED": SELECT_Q = f'''SELECT ls.id,ls.added::text, dm.name,dm.mobile, ls.bvn , ls.loan,lo.description,ls.amount,ls.verified::text, 200 AS total_fee FROM loan_apply ls LEFT JOIN loan_offers lo ON lo.loan=ls.loan LEFT JOIN demo_bank_accounts dm ON dm.bvn = ls.bvn WHERE ls.verified IS NOT NULL AND ls.status = 5 ORDER BY ls.id DESC LIMIT 300''' if loanLevel == "DISBURSEMENTS": SELECT_Q = f'''SELECT ls.id,ls.added::text, dm.name, ls.bvn , ls.loan,lo.description,l.approved_amount,ls.verified::text, 200 AS total_fee, l.due_date::text, (CASE WHEN l.status = 1 THEN 'Pending Transmission' ELSE 'Unknown' END) AS transmission_status FROM loans l LEFT JOIN loan_apply ls ON ls.uid::text = l.application_uid LEFT JOIN loan_offers lo ON lo.loan=ls.loan LEFT JOIN demo_bank_accounts dm ON dm.bvn = ls.bvn WHERE ls.verified IS NOT NULL AND ls.status = 5 ORDER BY ls.id DESC LIMIT 300''' with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_Q) select_demoS = cursor.fetchall() demo_data = json.dumps( [dict(ix) for ix in select_demoS] ) return json.loads( demo_data) @app.route('/office/offers') def offer_offers(): OFFER_QUERY = "SELECT uid AS cid,loan,amount,description,days_duration,active,score,lorder FROM loan_offers " with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(OFFER_QUERY) select_demoS = cursor.fetchall() loan_data = json.dumps( [dict(ix) for ix in select_demoS] ) loan_result = json.loads( loan_data) return loan_result @app.route('/loan/info') def pending_loans(): offer_detail = [ {"loan_id": "425611f2-c692-4404-b93d-76ca7a5ce00", "description": "Total amount to replay N50,000" , "active" : 1 }, {"loan_id": "425611f2-c692-4404-b93d-76ca7a5ce01", "description": "Total amount to replay N150,000" , "active" : 1 }, {"loan_id": "425611f2-c692-4404-b93d-76ca7a5ce02", "description": "Total amount to replay N300,000" , "active" : 1 }, ] return offer_detail @app.route('/loan/repay', methods=["POST"]) def repay_loans(): # try: # data = request.json # if not data: # return { # "message": "Please provide pin,loan_id,accountID", # "data": None, # "error": "Bad request" # }, 400 # validate input # bvn = data["bvn"] # accountid = data["accountid"] # loan_id = data["bvn"] offer_detail = [ {"cid": "425611f2-c692-4404-b93d-76ca7a5ce00", "description": "Total amount to replay N50,000" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce01", "description": "Total amount to replay N150,000" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce02", "description": "Total amount to replay N300,000" , "active" : 1 }, ] return offer_detail @app.route('/simulation/generator') def generate_activities(): # insert_random_users() generate_simulation_salary() generate_simulation_transaction() generate_simulation_credit() return [] def insert_random_users(): SQL_ACC = "SELECT accountid AS account_no FROM trx_raw GROUP BY accountid" with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_ACC) select_demoS = cursor.fetchall() loan_data = json.dumps( [dict(ix) for ix in select_demoS] ) accounts_result = json.loads( loan_data) SQL_INSERT = "INSERT INTO demo_bank_accounts (name, salary_account, mobile, bvn, email, pin) VALUES(%s,%s,%s,%s,%s,%s)" # Iterate through the JSON array for item in accounts_result: accountNo = item["account_no"] new_data = ( names.get_full_name() , random.randint(0, 1), '811-'+str( random.randint(100, 999) )+'-'+str( random.randint(1000, 9999) ) , accountNo, 'demo+'+ names.get_first_name() +'@chiefsoft.net', random.randint(1100, 9999) ) print(new_data) with connection: try: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_INSERT, new_data) except Exception as e: print(0) def generate_simulation_salary(): pass def generate_simulation_transaction(): pass def generate_simulation_credit(): pass def refresh_accounts(): SQL_ACC_f ="SELECT accountid,counters FROM customer_account_list WHERE counters > 10 AND created IS NULL LIMIT 5000" print(SQL_ACC_f) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_ACC_f) select_demoS = cursor.fetchall() account_data = json.dumps( [dict(ix) for ix in select_demoS] ) accounts_result = json.loads( account_data) trx_start_date = date.today() date_format = "%d-%m-%Y" for item in accounts_result: accountid = item["accountid"].strip() counters = item["counters"] days_to_add = 365*random.randint(2,15) + counters * 28 + random.randint(0, 5) days_to_created = trx_start_date - timedelta(days=days_to_add) SQLUP = "UPDATE customer_account_list SET created = '" + str(days_to_created) + "' WHERE accountid ='" + accountid + "'" print(SQLUP) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQLUP) return [] @app.route('/transaction/generator') def generate_transactions(): refresh_accounts() # SQL_ACC = "SELECT id, accountid,counters FROM customer_account_list WHERE counters > 0 AND counters < 14 ORDER by counters, id ASC LIMIT 700" SQL_ACC = "SELECT id, accountid,counters FROM customer_account_list WHERE counters > 0 AND counters < 14 ORDER by counters, id ASC LIMIT 700" with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_ACC) select_demoS = cursor.fetchall() account_data = json.dumps( [dict(ix) for ix in select_demoS] ) accounts_result = json.loads( account_data) # Iterate through the JSON array for item in accounts_result: accountid = item["accountid"].strip() counters = item["counters"] individual_transactions(accountid,counters) return [] def individual_transactions(accountid,counters): SQL_UPDATE_COUNTERS = "UPDATE customer_account_list SET counters = counters + 1 WHERE trim(accountid) = '" + accountid + "' " print(SQL_UPDATE_COUNTERS) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_UPDATE_COUNTERS) SQL_TRX_MODEL = "SELECT accountid, trx_start_date, trx_end_date, amount, d1, d2 , d3, description, d4 FROM trx_raw WHERE accountid='" + accountid + "'" print(SQL_TRX_MODEL) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_TRX_MODEL) select_demoS = cursor.fetchall() account_data = json.dumps( [dict(ix) for ix in select_demoS] ) strx_result = json.loads( account_data) # Iterate through the JSON array for item in strx_result: accountid = item["accountid"] trx_start_date = item["trx_start_date"].strip() trx_end_date = item["trx_end_date"].strip() date_string = "2025-03-28" date_format = "%d-%m-%Y" days_to_add = 365*6 + counters * 28 + random.randint(0, 5) trx_start_date_cal = datetime.strptime(trx_start_date, date_format).date() + timedelta(days=days_to_add) trx_end_date_cal = trx_start_date_cal amount = item["amount"] * (1 + random.randint(0, 10)/100) d1 = item["d1"].strip() d2 = item["d2"].strip() d3 = item["d3"].strip() description = item["description"].strip() d4 = item["d4"].strip() new_data = ( accountid, trx_start_date_cal, trx_end_date_cal, amount, d1, d2 , d3, description, d4 ) print(new_data) SQL_INSERT = "INSERT INTO customer_account_transaction_hx (accountid, trx_start_date, trx_end_date, amount, d1, d2 , d3, description, d4) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s)" print(SQL_INSERT) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SQL_INSERT,new_data) return [] if __name__ == '__main__': app.run(host='0.0.0.0', port=8000)