1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38
| import sqlite3
from contextlib import closing
def show_records(cnx, state):
sql = """select * from test where state = :state"""
with closing(cnx.cursor()) as c:
c.execute(sql, vars())
rows = c.fetchall()
print ('state: %s, count= %d' % (state, len(rows)))
def update_records(cnx, value, state):
sql = "update test set value = :value where state = :state"
with closing(cnx.cursor()) as c:
c.execute(sql, vars())
print ('state: %s, updated: %d' % (state, c.rowcount))
if __name__ == '__main__':
import random
ddl = """create table test(
name text primary key,
state text,
value integer default 1)"""
cnx = sqlite3.connect(':memory:')
cnx.execute(ddl)
COUNT = 10
STATES = ('a', 'b', 'c', 'd')
records = [ ('ident-%d' % x, random.choice(STATES)) for x in range(COUNT) ]
cnx.executemany('insert into test (name, state) values (:name, :state)', records)
show_records(cnx, 'a')
update_records(cnx, 2, 'a')
show_records(cnx, 'a')
show_records(cnx, 'x')
update_records(cnx, 2, 'x') |
Partager