Load data using SQL loader via shell or equivalent python

---------------------------------------------------------------------

Shell script:

==========run_sqlldr.sh==============

#!/bin/bash
# Oracle Environment
export ORACLE_HOME=/u01/app/oracle/product/19.0.0/db_1
export PATH=$ORACLE_HOME/bin:$PATH
export ORACLE_SID=oradb
# Variables
USERID=ashwani/ashwani@pdb1
CONTROL_FILE=/home/oracle/emp.ctl
LOG_FILE=/home/oracle/emp.log
BAD_FILE=/home/oracle/emp.bad
DATA_FILE=/home/oracle/emp.csv
echo "Starting SQL Loader..."
sqlldr userid=$USERID \
control=$CONTROL_FILE \
data=$DATA_FILE \
log=$LOG_FILE \
bad=$BAD_FILE
if [ $? -eq 0 ]
then
   echo "Data loaded successfully."
else
   echo "SQL Loader failed."
fi

------------------------------------------------------------------------

Python script:

=============run_sqlldr.py============================================

#!/usr/bin/env python3

import argparse

import os

import subprocess

import sys

from datetime import datetime

# ---------------------------------------------------------------------------

# Oracle environment

# ---------------------------------------------------------------------------

ORACLE_HOME = "/u01/app/oracle/product/19.0.0/db_1"

ORACLE_SID  = "oradb"

# ---------------------------------------------------------------------------

# Default SQL*Loader parameters

# ---------------------------------------------------------------------------

DEFAULT_USERID   = "ashwani/ashwani@pdb1"

DEFAULT_CONTROL  = "/home/oracle/emp.ctl"

DEFAULT_LOG      = "/home/oracle/emp.log"

DEFAULT_BAD      = "/home/oracle/emp.bad"

DEFAULT_DATA     = "/home/oracle/emp.csv"

def parse_args():

    p = argparse.ArgumentParser(description="Python wrapper for SQL*Loader (sqlldr).")

    p.add_argument("--oracle-home", default=ORACLE_HOME)

    p.add_argument("--oracle-sid",  default=ORACLE_SID)

    p.add_argument("--userid",      default=DEFAULT_USERID)

    p.add_argument("--control",     default=DEFAULT_CONTROL)

    p.add_argument("--data",        default=DEFAULT_DATA)

    p.add_argument("--log",         default=DEFAULT_LOG)

    p.add_argument("--bad",         default=DEFAULT_BAD)

    return p.parse_args()

def build_env(oracle_home, oracle_sid):

    env = os.environ.copy()

    env["ORACLE_HOME"] = oracle_home

    env["ORACLE_SID"]  = oracle_sid

    env["PATH"]        = os.path.join(oracle_home, "bin") + os.pathsep + env.get("PATH", "")

    lib = os.path.join(oracle_home, "lib")

    env["LD_LIBRARY_PATH"] = lib + os.pathsep + env.get("LD_LIBRARY_PATH", "")

    return env

def main():

    args = parse_args()

    env  = build_env(args.oracle_home, args.oracle_sid)

    sqlldr = os.path.join(args.oracle_home, "bin", "sqlldr")

    cmd = [

        sqlldr,

        "userid="  + args.userid,

        "control=" + args.control,

        "data="    + args.data,

        "log="     + args.log,

        "bad="     + args.bad,

    ]

    print("Starting SQL Loader...")

    print("Command:", " ".join(cmd))

    # universal_newlines=True is the Python 3.6-compatible equivalent of text=True

    result = subprocess.run(cmd, env=env,

                            stdout=subprocess.PIPE,

                            stderr=subprocess.STDOUT,

                            universal_newlines=True)

    print(result.stdout or "")

    # Append run metadata to log

    try:

        with open(args.log, "a") as lf:

            lf.write("\n" + "="*60 + "\n")

            lf.write("Python runner: {}\n".format(datetime.now()))

            lf.write("Return code : {}\n".format(result.returncode))

    except OSError as e:

        print("WARNING: could not write log - {}".format(e))

    if result.returncode == 0:

        print("Data loaded successfully.")

    else:

        print("SQL Loader failed. (exit code: {})".format(result.returncode))

        sys.exit(result.returncode)

if __name__ == "__main__":

    main()

=============

create table:

=============

CREATE TABLE emp (

    emp_id NUMBER GENERATED ALWAYS AS IDENTITY,

    emp_name VARCHAR2(100),

    salary NUMBER,

    created_time TIMESTAMP DEFAULT SYSTIMESTAMP

);

===================

Data file: cat emp.csv

===================

PETER,50000

RICHARD,60000

ASHWANI,75000

================

Control File: emp.ctl

================

LOAD DATA

INFILE 'emp.csv'

INSERT

INTO TABLE emp

FIELDS TERMINATED BY ','

TRAILING NULLCOLS

(

  emp_name,

  salary

)