Updating and reading binary data

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