Das deutsche Python-Forum

Code: Alles auswählen

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import cgitb;cgitb.enable()

print "Content-type: text/html; charset=utf-8\r\n"

import adodbapi as dbapi


DSN_info = (
    "DRIVER=SQL Server;"
    "SERVER=.\SQLEXPRESS;"
    "DATABASE=DatabaseName;"
    "UID=UserName;"
    "PASSWORD=Password;"
)


class test:
    def __init__( self ):
        print "<pre>"
        self.connect()
        self.make_test_table()
        self.select_all()

        self.insert_1()
        self.select_all()

        self.delete_1()
        self.select_all()

        self.delete_test_table()
        self.close()
        print "</pre>"

    def connect(self):
        print "connect:",
        print DSN_info

        try:
            self.connection = dbapi.connect(DSN_info)
            self.cursor = self.connection.cursor()
        except Exception, e:
            print "Error:", e
            sys.exit()
        else:
            print "Connection OK"

    def close(self):
        self.connection.close()
        print "\n\nConnection closed"

    def make_test_table(self):
        SQLcommand  = (
            "CREATE TABLE TestTable ("
            "id INTEGER NOT NULL PRIMARY KEY IDENTITY(1,1), "
            "data1 VARCHAR( 50 ) NOT NULL, "
            "data2 VARCHAR( 50 ) NOT NULL"
            ");"
        )

        self.execute(
            "Creat a temporary test table",
            SQLcommand
        )

    def delete_test_table(self):
        self.execute(
            "Delete temporary test table",
            "DROP TABLE TestTable;"
        )

    def insert_1(self):
        self.execute(
            "RAW INSERT without SQLescaping",
            "INSERT INTO TestTable (data1,data2) VALUES ('Value A 1', 'Value A 2');"
        )

    def select_all(self):
        result = self.fetchall(
            "SELECT all",
            "SELECT * FROM TestTable;"
        )
        print result

    def delete_1(self):
        self.execute(
            "RAW DELETE values",
            "DELETE FROM TestTable WHERE data1='Value A 1';"
        )

    def execute(self, comment, SQLcommand):
        print "\n\n%s:" % comment
        print "SQLcommand:", SQLcommand
        try:
            self.cursor.execute(SQLcommand)
        except Exception, e:
            print "Error:", e
        else:
            print "OK"

    def fetchall(self, comment, SQLcommand):
        print "\n\n%s:" % comment
        print "SQLcommand:", SQLcommand
        try:
            self.cursor.execute(SQLcommand)
            return self.cursor.fetchall()
        except Exception, e:
            print "Error:", e
        else:
            print "OK"



if __name__ == "__main__":
    test()

Code: Alles auswählen

if self.supportsTransactions:
    self.adoConn.IsolationLevel=defaultIsolationLevel
    self.adoConn.BeginTrans() #Disables autocommit