Updating and reading binary data

Updating and reading binary data
  1# This script shows how to use the IDatabaseQuery object to read BLOB data from a row and use an
  2# IDatabaseCommand object to update BLOB data in a table using an SQL update statement.
  3
  4
  5BLOBSourceFile = r"D:\\image.jpg"
  6BLOBDestinationFile = r"D:\\image2.jpg"
  7
  8ui = impact.gui
  9ot = ui.output_toolbox
 10
 11db = impact.active_database
 12
 13ot.clear()
 14
 15ad_type_binary = 1
 16ad_save_create_over_write = 2
 17
 18
 19# IDatabaseConnection for current impact database
 20c = db.connection
 21
 22
 23def read_binary_file(file_name):
 24
 25    # create stream object
 26    # stream = CreateObject("ADODB.Stream")
 27
 28    # load the file data from disk to stream object
 29    stream.open()
 30    stream.type = ad_type_binary
 31    stream.load_from_file(file_name)
 32
 33    # get binary data from the object
 34    read_binary_file = stream.read
 35
 36def save_binary_data(file_name, byte_array):
 37
 38    # stream = CreateObject("ADODB.Stream")
 39
 40    # open the stream and write binary data to the object
 41    stream.open()
 42    stream.type = ad_type_binary
 43
 44    # write binary data to the file
 45    stream.write(byte_array)
 46    stream.save_to_file(file_name, ad_save_create_over_write)
 47
 48def update_test_value():
 49    cmd = c.create_command()
 50
 51    # assign the SQL including parameter placeholders
 52    cmd.sql = "update TEST set T_DATA = :data where T_KEY = :key"
 53
 54    # assign parameters to command
 55    cmd.parameter("data").value = read_binary_file(BLOBSourceFile)
 56    cmd.parameter("key").value = 1
 57
 58    # when executing a simple SQL statement you don't need a transaction - one will automatically be used
 59    if cmd.execute_sql():
 60        ot.add("Successfully updated record")
 61
 62    else:
 63        ot.add("Unable to execute command")
 64
 65def read_test_value():
 66    q = c.create_query()
 67
 68    q.sql = "select T_DATA, " + str(c.blob_length("T_DATA")) + " from TEST where T_KEY = :key"
 69    q.parameter("key").value = 1
 70
 71    if q.open():
 72        if not q.is_eof:
 73            blob_c = q.column("T_DATA")
 74            blob_len_c = q.column(2)
 75
 76            if blob_c.is_null:
 77                ot.add("TDATA=NULL")
 78            else:
 79                ot.add("TDATA BLOB Length=" + str(blob_len_c.value))
 80
 81                # we need the BLOB value as a VARIANT array of bytes
 82                blob_c.bytes_as_variant = False
 83
 84                save_binary_data(BLOBDestinationFile, blob_c.value)
 85
 86                ot.add("Successfully saved BLOB to '" + str(BLOBDestinationFile) + "'")
 87
 88        else:
 89            ot.add("Unable to locate T_TEST record")
 90
 91        if not q.close():
 92            ot.add("Unable to close query")
 93
 94    else:
 95        ot.add("Unable to open query")
 96
 97
 98ot.clear()
 99
100ot.add("Connection Name: " + str(c.connection_name))
101ot.add("Connection Type: " + str(c.connection_type))
102ot.add("DBMS Type: " + str(c.dbms_type))
103ot.add("DBMS version: " + str(c.dbms_version.as_string))
104
105# save BLOB into TEST table
106UpdateTESTValue()
107
108# read BLOB from TEST table
109ReadTESTValue()