DB Testing via ODBC

Using Windows ODBC with Python

Note this script explicitly references the ActiveState binary so as to utilize its in built ODBC module.

This script draws the required Manifest Ids from the Data Server (AKA dserver) which is described elsewhere.

Getting ODBC running with Python under Cygwin had too many external dependencies to make it a simple job. It is still on the agenda but will have to wait for the moment.

#!/c/PROGRA~1/ActiveState/Python/python.exe
#
#-------------------------------------------------------------------------------

import os
import sys
import dbi
import odbc
import time
import getopt
import logging

from datetime import datetime

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

from RTE import dcl

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

__version__   = "1.0.0"
__id__        = "@(#)  qry.py  [%s]  05/03/2007"

verbose_flg   = False

debug_level   = 0


table_name    = "Manifests"
PORT          = 9579
MAX_REQUESTS  = 10

log           = None
handle        = 'xyzzy_%02d'
connector_no  = 1

#===============================================================================

def INFO(msg):
   if log: log.info(' ' + msg)
   if verbose_flg: print "[test]  %s" % msg

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

def ERROR(msg):
   if log: log.error(msg)
   sys.stderr.write('[test]  %s\n' % msg)

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

def WARNING(msg):
   if log: log.warning('*****' + msg + '*****')
   if verbose_flg: print "[test]  %s" % msg

#===============================================================================

def the_time():
   t = float(ref_time(False)) * 0.001

   return t

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

t_reference = None

def ref_time(flg):
   global t_reference

   t_now  = datetime.now()

   if (flg):
      t_reference   = t_now
      t             = 0
   else:
      t_delta       = t_now - t_reference
      t             = ((t_delta.seconds * 1000000) + t_delta.microseconds)/1000.0

   return t

#===============================================================================

def query(no_queries, connector_no):
   ds = dcl.dcl(port=PORT)

   if (ds == None):
      print("Connection to data server failed - is data server process running?\n")
      return 1

   type_ref  = ds.RegisterType(table_name)

   if debug_level > 0:print "Data type \"%s\" registered as %d" % (table_name,  type_ref)

   s = odbc.odbc(handle % connector_no)

   cur = s.cursor()

   cnt = 0

   while cnt < no_queries:
      cnt += 1

      sp  = ds.GetNext(type_ref)

      if sp != None:
         manifest = sp[0]
      else:
         print "Type %d exhausted" % (pid, type_ref)
         return

      # print manifest

      qry   = "exec dbp_get_Article_history_for_internet %s" % manifest

      # print qry

      ref_time(True)

      cur.execute(qry)

      t_query = the_time()

      # print cur.description

      # for tup in cur.description:
         # print tup[0]

      row_cnt = 0

      while 1:
         rec = cur.fetchmany(10)
         row_cnt +=  len(rec)
         if not rec: break
         #print rec

      msg = "%02d ManifestNo %s  [%.3f] sec  %3d rows returned" % (cnt, manifest, t_query, row_cnt)

      sys.stderr.write("%s\n" % msg)
      sys.stderr.flush()

      INFO(msg)

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

def init():
   global lf
   global log

   pid      = os.getpid()

   if debug_level > 0:  print "My PID is %d" % pid

   log  = logging.getLogger('test')
   hdlr = logging.FileHandler('log/hcodevj86_%02d.log' % connector_no)
   fmtr = logging.Formatter('%(asctime)s %(levelname)s %(message)s')

   hdlr.setFormatter(fmtr)
   log.addHandler(hdlr) 
   log.setLevel(logging.INFO)

   INFO("===== Started processing ==================================")

#===============================================================================

def main():
   global verbose_flg
   global debug_level
   global connector_no
   global MAX_REQUESTS

   try:
      opts, args = getopt.getopt(sys.argv[1:], "c:dD:M:vVw?")
   except getopt.error, msg:
      print __doc__
      return 1

   for o, a in opts:
      if o == '-?':
         print __doc__
         return 1
      elif o == '-c':
         connector_no   = int(a)
      elif o == '-d':
         debug_level   += 1
      elif o == '-D':
         debug_level    = int(a)
      elif o == '-M':
         MAX_REQUESTS   = int(a)
      elif o == '-v':
         verbose_flg    = True
      elif o == '-V':
         print "[xxxx]  Version: %s" % __version__
         return 1
      else:
         print __doc__
         return 1

   sys.stderr.write("[test]  Working directory is %s - Using ODBC connector %d\n" % (os.getcwd(), connector_no))
   sys.stderr.flush()

   if (debug_level > 0): print "Debugging level set to %d" % debug_level

   if args:
      for arg in args:
         print arg

   init()

   query(MAX_REQUESTS, connector_no)

   return 0

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

if __name__ == '__main__' or __name__ == sys.argv[0]:
   try:
      sys.exit(main())
   except KeyboardInterrupt, e:
      print "[xxxx]  Interrupted!"

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

"""
Revision History:

     Date     Who   Description
   --------   ---   ------------------------------------------------------------
   20031014   plh   Initial implementation

Problems to fix:

To Do:

Issues:

"""

LoadRunner Script

Here is the LoadRunner equivalent of the above python script without the 'bells and whistles' we would normally add to complete the script.

vuser_init.c

#include "vdf.h"
#include "print.inl"

vuser_init()
{
    lrd_init(&InitInfo, DBTypeVersion);

    lrd_open_context(&Ctx1, LRD_DBTYPE_ODBC, 0, 0, 0);
    lrd_alloc_connection(&Con1, LRD_DBTYPE_ODBC, Ctx1, 0 /*Unused*/, 0);
    lrd_db_option(Con1, OT_ODBC_LOGIN_TIMEOUT, (void FAR const *)30, 0);

    lrd_open_connection(&Con1, LRD_DBTYPE_ODBC, "", lr_decrypt("xxxxx"), "",
        lr_decrypt("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"), Ctx1, 1, 0);

    return VUSER_OK;
  // vuser_init

Action.c

Action()
{
    lrd_open_cursor(&Csr12, Con1, 0);

    lrd_db_option(Csr12, OT_ODBC_CURSOR_FWD_ONLY, 0, 0);
    lrd_db_option(Csr12, OT_ODBC_CONCUR_READ_ONLY, 0, 0);
    lrd_db_option(Csr12, OT_ODBC_MAX_ROWS, (void FAR const *)100, 0);
    lrd_db_option(Csr12, OT_ODBC_CURSOR_CLOSE, 0, 0);

    lr_start_transaction("GetArticleHistory_stmt");

    lrd_stmt(Csr12, "exec dbp_get_Article_history_for_internet {ManifestId}",
        -1, 1 /*Direct exec*/, 0 /*None*/, 0);

    lr_end_transaction("GetArticleHistory_stmt", LR_AUTO);

    lr_start_transaction("GetArticleHistory_fetch");

    lrd_fetch(Csr12, -11, 1, 0, PrintRow12, 0);
    // GRID(12);

    lr_end_transaction("GetArticleHistory_fetch", LR_AUTO);

    lrd_result_set(Csr12, 0, 0, 0);
    lrd_close_cursor(&Csr12, 0);

    return VUSER_OK;
}

vuser_end.c

vuser_end()
{
    lrd_close_connection(&Con1, 0, 0);

    lrd_free_connection(&Con1, 0 /*Unused*/, 0);

    lrd_close_context(&Ctx1, 0, 0);

    lrd_end(0);

    return VUSER_OK;
}  // vuser_end