odb_sqlite.py

在 linux平台上的网页编程的模板

#! /usr/bin/env python

"""
usage: %(progname)s [args]
"""


import os, sys, string, time, getopt
from log import *

import odb
import sqlite

import re

# --- these are using for removing nulls from strings
# --- because sqlite can't handle them

def escape_string(str):
    def subfn(m):
        c = m.group(0)
        return "%%%02X" % ord(c)

    return re.sub("('|\0|%)",subfn,str)

def unescape_string(str):
    def subfn(m):
        hexnum = int(m.group(1),16)
        return "%c" % hexnum
    return re.sub("%(..)",subfn,str)

class Database(odb.Database):
  def __init__(self,db, debug=0):
    odb.Database.__init__(self, db, debug=debug)
    self.SQLError = sqlite.Error
    
  def escape(self,str):
    if str is None:
      return None
    elif type(str) == type(""):
      return string.replace(str,"'","''")
    elif type(str) == type(1):
      return str
    else:
      raise "unknown column data type: %s" % type(str)


  def listTables(self, cursor=None):
    if cursor is None: cursor = self.defaultCursor()
    cursor.execute("select name from sqlite_master where type='table'")
    rows = cursor.fetchall()
    tables = []
    for row in rows: tables.append(row[0])
    return tables

  def listIndices(self, cursor=None):
    if cursor is None: cursor = self.defaultCursor()
    cursor.execute("select name from sqlite_master where type='index'")
    rows = cursor.fetchall()
    tables = []
    for row in rows: tables.append(row[0])
    return tables

  def listFieldsDict(self, table_name, cursor=None):
    if cursor is None: cursor = self.defaultCursor()
    sql = "pragma table_info(%s)" % table_name
    cursor.execute(sql)
    rows = cursor.fetchall()

    columns = {}
    for row in rows:
      colname = row[1]
      columns[colname] = row
    return columns

  def _tableCreateStatement(self, table_name, cursor=None):
    if cursor is None: cursor = self.defaultCursor()
    sql = "select sql from sqlite_master where type='table' and name='%s'" % table_name
    print sql
    cursor.execute(sql)
    row = cursor.fetchone()
    sqlstatement = row[0]
    return sqlstatement
    

  def alterTableToMatch(self, table):
    tableName = table.getTableName()
    tmpTableName = tableName + "_" + str(os.getpid())


    invalidAppCols, invalidDBCols = table.checkTable(warnflag=0)

##     if invalidAppCols or invalidDBCols:
##       return

    if not invalidAppCols and not invalidDBCols:
      return


    oldcols = self.listFieldsDict(tableName)
#    tmpcols = oldcols.keys()
    
    tmpcols = []
    newcols = table.getAppColumnList()
    for colname, coltype, options in newcols:
      if oldcols.has_key(colname): tmpcols.append(colname)
    
    tmpcolnames = string.join(tmpcols, ",")
      
    statements = []

    sql = "begin transaction"
    statements.append(sql)

    sql = "create temporary table %s (%s)" % (tmpTableName, tmpcolnames)
    statements.append(sql)

    sql = "insert into %s select %s from %s" % (tmpTableName, tmpcolnames, tableName)
    statements.append(sql)

    sql = "drop table %s" % tableName
    statements.append(sql)
    
    sql = table._createTableSQL()
    statements.append(sql)

    sql = "insert into %s(%s) select %s from %s" % (tableName, tmpcolnames, tmpcolnames, tmpTableName)
    statements.append(sql)

    sql = "drop table %s" % tmpTableName
    statements.append(sql)

    sql = "commit"
    statements.append(sql)

    cur = self.defaultCursor()
    for statement in statements:
#      print statement
      cur.execute(statement)


def test():
  pass

def usage(progname):
  print __doc__ % vars()

def main(argv, stdout, environ):
  progname = argv[0]
  optlist, args = getopt.getopt(argv[1:], "", ["help", "test", "debug"])

  testflag = 0
  if len(args) == 0:
    usage(progname)
    return
  for (field, val) in optlist:
    if field == "--help":
      usage(progname)
      return
    elif field == "--debug":
      debugfull()
    elif field == "--test":
      testflag = 1

  if testflag:
    test()
    return


if __name__ == "__main__":
  main(sys.argv, sys.stdout, os.environ)