dbtest.py

python web programming 部分

"""test script for database connectivity

usage dbtest.py <directory>

run in current directory creates a database in files
  test.dfs LIKES.grl SERVES.grl FREQUENTS.grl
"""
def cpp(cursor, t = None):
    print
    d = cursor.description
    if not d:
        print "#### NO RESULTS ###\n"
        return
    names = []
    lengths = []
    rules = []
    for dd in d:
        l = dd[1]
        if not l:
            l = 12
        l = max(l, len(dd[0]))
        names.append(dd[0])
        lengths.append("%%%ss" % l)
        rules.append("-"*l)
    format = " ".join(lengths)
    print format % tuple(names)
    print format % tuple(rules)
    if not t:
        t = cursor.fetchall()
    for row in t:
        print format % row
    print

def test():
    import sys
    print "testing"
    connect = dbconnect()
    curs = connect.cursor()
    for q in cleanup_queries:
        try:
            print "Cleanups:", q
            sys.stdout.flush()
            curs.execute(q)
            sys.stdout.flush()
        except: # this is a bit dodgy: don't know error classes
            pass
    print
    print "TABLE CREATES"
    for x in table_creates:
        print x
        curs.execute(x)
    curs.execute("create table empty (nothing varchar)")
    C = """
    CREATE TABLE workt (
       name VARCHAR(10),
       hours INTEGER,
       rate FLOAT)
       """
    print C
    curs.execute(C)
    print
    C = """
    CREATE TABLE pagerefs (
       page VARCHAR(20),
       hits INTEGER,
       month INTEGER)
       """
    print C
    curs.execute(C)
    print
    print "INSERTS"
    C = """
    INSERT INTO workt(name, hours, rate) VALUES (?, ?, ?)
    """
    D = [
         ("sam", 30, 40.2),
         ("norm", 45, 10.2),
         ("woody", 80, 5.4),
         ("diane", 3, 4.4),
         ("rebecca", 120, 12.9),
         ("cliff", 26, 200.00),
         ("carla", 9, 3.5),
         ]
    for x in D: print x
    curs.execute(C, D)
    C = "create unique index wname on workt(name)"
    print "Unique index:", C
    curs.execute(C)
    print "trying bad insert into unique field"
    C = "insert into workt(name, hours, rate) values ('sam', 0, 0)"
    try:
        curs.execute(C)
    except:
        print "exception as expected %s(%s)" %(sys.exc_type, sys.exc_value)
    else:
        raise "stop!", "unique index permits nonunique field"
    C = """
    INSERT INTO pagerefs(page, month, hits) VALUES (?, ?, ?)
    """
    D = [
         ("index.html", 1, 2100),
         ("index.html", 2, 3300),
         ("index.html", 3, 1950),    
         ("products.html", 1, 15),   
         ("products.html", 2, 650),   
         ("products.html", 3, 98),   
         ("people.html", 1, 439),
         ("people.html", 2, 12),
         ("people.html", 3, 665),
         ]
    for x in D: print x
    curs.execute(C, D)
    for (table, stuff) in dpairs:
        ins = "insert into %s values (?, ?, ?)" % table
        if table!="frequents":
           for parameters in dataseq(stuff):
               print "singleinsert", table, parameters
               curs.execute(ins, parameters)
        else:
           print
           print "multiinsert", table
           parameters = dataseq(stuff)
           for p in parameters:
               print p
           print "multiinsert..."
           curs.execute(ins, parameters)
           print;print
    print
    print "INDICES"
    for ci in indices:
        print ci
        curs.execute(ci)
    print
    print "QUERIES"
    for x in workqueries:
        print;print
        print x
        t = curs.execute(x)
        cpp(curs, t)
        
    statement = """select name, hours
                   from workt"""
    curs.execute(statement)
    print "Hours worked this week"
    print
    for (name,hours) in curs.fetchall():
        print "worker", name, "worked", hours, "hours"
    print
    print "end of work report"

    #return
    for x in queries:
        print; print
        print x
        curs.execute(x)
        #for x in curs.commands:
        #    print x
        all = curs.fetchall()
        if not all:
           print "empty!"
        else:
           cpp(curs, all)
#           for t in all:
#               print t
    #return
    print
    print "DYNAMIC QUERIES"
    for (x,y) in dynamic_queries:
        print; print
        print x
        print "dynamic=", y
        curs.execute(x, y)
        #for x in curs.commands:
        #    print x
        all = curs.fetchall()
        if not all:
           print "empty!"
        else:
           for t in all:
               print t
    print "repeat test"
    from time import time
    for x in repeats:
        print "repeating", x
        now = time()
        curs.execute(x)
        print time()-now, "first time"
        now = time()
        curs.execute(x)
        print time()-now, "second time"
        now = time()
        curs.execute(x)
        print time()-now, "third time"
    print "*** committing work"
    connect.commit()
    connect.close()
    print; print
    print "*" * 30
    print "*** reconnect test"
    connect = reconnect()
    cursor = connect.cursor()
    for s in updates + cleanup_queries:
        print; print
        print s
        cursor.execute(s)
        cpp(cursor)
    connect.commit()
    connect.close()
    return connect


table_creates = [
 "create table frequents (drinker varchar(10), bar varchar(10), perweek integer)",
 "create table likes (drinker varchar(10), beer varchar(12), perday integer)",
 "create table serves (bar varchar(10), beer varchar(12), quantity integer)",
 """Create view nondrinkers(d, b)
    as select drinker, bar
       from frequents
       where drinker not in
         (select drinker from likes)""",
 ]
 
fdata = """\
  adam	lolas		1
  woody	cheers	5
  sam		cheers	5
  norm	cheers	3
  wilt	joes		2
  norm	joes		1
  lola	lolas		6
  norm	lolas		2
  woody	lolas		1
  pierre	frankies	0"""
  
sdata = """\
  cheers	bud		500
  cheers	samaddams	255
  joes	bud		217
  joes	samaddams	13
  joes	mickies	2222
  lolas	mickies	1515
  lolas	pabst		333
  winkos	rollingrock	432
  frankies	snafu       5"""
  
ldata = """\
  adam	bud			2
  wilt	rollingrock		1
  sam		bud			2
  norm	rollingrock		3
  norm	bud			2
  nan		sierranevada	1	
  woody	pabst			2
  lola	mickies		5"""
  
dpairs = [
   ("frequents", fdata),
   ("serves", sdata),
   ("likes", ldata),
  ]
  
def dataseq(s):
    from string import split
    l = split(s, "\n")
    result = map(split, l)
    from string import atoi
    for l in result:
        l[2] = atoi(l[2])
    result = map(tuple, result)
    return result
    
indices = [
"""create index fd on frequents (drinker)""",
"""create index sbb on serves (beer, bar)""",
"""create index lb on likes (beer)""",
"""create index fb on frequents (bar)""",
]

repeats = [
"""-- drinkers bars and beers
   -- where the drinker likes the beer
   -- the bar serves the beer
   -- and the drinker frequents the bar
   select f.drinker, l.beer, s.bar
   from frequents f, serves s, likes l
   where f.drinker=l.drinker and s.bar=f.bar and s.beer=l.beer""",
"""select *
   from frequents as f, serves as s
   where f.bar = s.bar and
     not exists(
       select l.drinker, l.beer
       from likes l
       where l.drinker=f.drinker and s.beer=l.beer)""",
"""select * from frequents
   where drinker = 'norm'""",
]

workqueries = [
"""-- everything from workt
   select * from workt""",
# stupid tests
"""select avg(hours)
   from workt""",
"""select *
   from workt
   where name='carla' -- just carla""",
"""select name, ' ain''t worth ', rate
   from workt -- the works table has more columns
   where name='carla'""",
"""select name, -- name of worker
          hours -- hours worked
   from workt""",
"""select name, hours*rate as pay
   from workt
   order by name""",
"""select name, rate
   from workt
   where rate>=20 and rate<=100""",
"""select name, rate
   from workt
   where rate between 20 and 100""",
"""select name, rate
   from workt
   where rate not between 20 and 100""",
"""select name, rate, hours, hours*rate as pay
   from workt""",
"""select name, rate, hours, hours*rate as pay
   from workt