sqlite3(python3.4 demo in doc)

import sqlite3

conn=sqlite3.connect('example.db')

cursor=conn.cursor()

cursor.execute('''create table stocks(date text,
trans text,
symbol text,
qty real,
price real)''')

<sqlite3.Cursor at 0x4867880>

cursor.execute("INSERT INTO stocks VALUES('2006-01-05','BUY','RHAT',100,35.14)")

<sqlite3.Cursor at 0x4867880>

conn.commit()

t=('RHAT',)

cursor.execute('SELECT * FROM  stocks WHERE symbol=?',t)

print(cursor.fetchone())

('2006-01-05', 'BUY', 'RHAT', 100.0, 35.14)

purchases=[('2006-03-28','BUY','IBM',1000,45.00),
('2006-04-05','BUY','MSFT',1000,72.00),
('2006-04-06','SELL','IBM',500,53.00)]

cursor.executemany('INSERT INTO stocks VALUES(?,?,?,?,?)',purchases)

[print(row) for row in cursor.execute('SELECT * FROM stocks ORDER BY price')]

('2006-01-05', 'BUY', 'RHAT', 100.0, 35.14)('2006-03-28', 'BUY', 'IBM', 1000.0, 45.0)
('2006-04-06', 'SELL', 'IBM', 500.0, 53.0)
('2006-04-05', 'BUY', 'MSFT', 1000.0, 72.0)

[None, None, None, None]

sqlite3.version

'2.6.0'

sqlite3.version_info

(2, 6, 0)

sqlite3.sqlite_version

'3.8.3.1'

sqlite3.sqlite_version_info

(3, 8, 3, 1)

import hashlib

def md5sum(t):
return hashlib.md5(t).hexdigest()

con=sqlite3.connect(':memory:')
con.create_function('md5',1,md5sum)# 1: number of parameter
cur=con.cursor()
cur.execute('select md5(?)',(b'foo',))#query the value of the md5sum(b'foo')
print(cur.fetchone()[0])

acbd18db4cc2f85cedef654fccc4a4d8

class MySum:
def __init__(self):
self.count=0

def step(self,value):
self.count+=value

def finalize(self):
return self.count

con.create_aggregate('mysum',1,MySum)

cur.execute('create table test(i)')
cur.execute('insert into test(i) values (1)')
cur.execute('insert into test(i) values (2)')

<sqlite3.Cursor at 0xa36030>

cur.execute('select mysum(i) from test')

<sqlite3.Cursor at 0xa36030>

print(cur.fetchone())

(3,)

cur.execute('select * from test')
cur.fetchall()

[(1,), (2,)]

def collate_reverse(str1,str2):
'''
The callable will be passed two string arguments.
It should return -1 if the first is ordered lower than the second,
0 if they are ordered equal and
1 if the first is ordered higher than the second.

Note that this controls sorting (ORDER BY in SQL)
so your comparisons don’t affect other SQL operations.

Note that the callable will get its parameters as Python bytestrings,
which will normally be encoded in UTF-8.

'''
if str1 == str2:
return 0;
elif str1<str2:
return 1
else:#str1 > str2
return -1

con.create_collation("reverse", collate_reverse)

cur.executemany("insert into test(i) values (?)", [("a",), ("b",)])

cur.execute("select i from test order by i collate reverse")
for row in cur:
print(row)

(1,)
(2,)
('b',)
('b',)
('b',)
('a',)
('a',)
('a',)

con.create_collation("reverse", None)

sorted(['a','b'])

['a', 'b']

# enable extension loading
con.enable_load_extension(True)

con.load_extension('./fts3.so')

---------------------------------------------------------------------------

OperationalError                          Traceback (most recent call last)

<ipython-input-52-bd51e5e015bb> in <module>()
----> 1 con.load_extension('./fts3.dll')

OperationalError: 找不到指定的模块。

def dict_factory(cursor,row):
d={}
for idx,col in enumerate(cursor.description):
d[col[0]]=row[idx]
return d

con.row_factory=dict_factory
cur.execute('select 1 as a')
print(cur.fetchone())

---------------------------------------------------------------------------

TypeError                                 Traceback (most recent call last)

<ipython-input-57-c8666e5801f2> in <module>()
7 con.row_factory=dict_factory
8 cur.execute('select 1 as a')
----> 9 print(cur.fetchone()['a'])

TypeError: tuple indices must be integers, not str

cur.description

(('a', None, None, None, None, None, None),)

AUSTRIA='\xd6sterreich'

# by default, rows are returned as Unicode
cur.execute("select ?", (AUSTRIA,))

<sqlite3.Cursor at 0xa36030>

row=cur.fetchone()

assert row[0] ==AUSTRIA

con.text_factory=bytes
cur.execute('select ?',(AUSTRIA,))
row=cur.fetchone()
assert type(row[0]) is bytes

assert row[0]==AUSTRIA.encode('utf-8')
con.text_factory=lambda x:x.decode('utf-8')+'foo'
cur.execute('select ?',('bar',))
row=cur.fetchone()
assert row[0]=='barfoo'

who='Yeltsin'
age=72
cur.execute("create table people (name_last, age)")

cur.execute('insert into people values(?,?)',(who,age))
cur.execute('select * from people where name_last=:who and age=:age',
{'who':who,'age':age})

<sqlite3.Cursor at 0xa36030>

print(cur.fetchone())

('Yeltsinfoo', 72)

class IterChars:
def __init__(self):
self.count=ord('a')

def __iter__(self):
return self

def __next__(self):
if self.count > ord('z'):
raise StopIteration
self.count+=1
return (chr(self.count-1),)

cur.execute('create table characters(c)')

<sqlite3.Cursor at 0xa36030>

theIter=IterChars()
cur.executemany('insert into characters(c) values(?)',theIter)

cur.execute('select c from characters')

print(cur.fetchall())

[('afoo',), ('bfoo',), ('cfoo',), ('dfoo',), ('efoo',), ('ffoo',), ('gfoo',), ('hfoo',), ('ifoo',), ('jfoo',), ('kfoo',), ('lfoo',), ('mfoo',), ('nfoo',), ('ofoo',), ('pfoo',), ('qfoo',), ('rfoo',), ('sfoo',), ('tfoo',), ('ufoo',), ('vfoo',), ('wfoo',), ('xfoo',), ('yfoo',), ('zfoo',)]

import string
string.ascii_lowercase

'abcdefghijklmnopqrstuvwxyz'

def char_generator():
for c in string.ascii_lowercase:
yield(c,)

iterr=char_generator()

next(iterr)

('a',)

cur.executemany("insert into characters(c) values (?)", char_generator())

<sqlite3.Cursor at 0xa36030>

cur.execute('select c from characters')
print(cur.fetchall())

[('afoo',), ('bfoo',), ('cfoo',), ('dfoo',), ('efoo',), ('ffoo',), ('gfoo',), ('hfoo',), ('ifoo',), ('jfoo',), ('kfoo',), ('lfoo',), ('mfoo',), ('nfoo',), ('ofoo',), ('pfoo',), ('qfoo',), ('rfoo',), ('sfoo',), ('tfoo',), ('ufoo',), ('vfoo',), ('wfoo',), ('xfoo',), ('yfoo',), ('zfoo',), ('afoo',), ('bfoo',), ('cfoo',), ('dfoo',), ('efoo',), ('ffoo',), ('gfoo',), ('hfoo',), ('ifoo',), ('jfoo',), ('kfoo',), ('lfoo',), ('mfoo',), ('nfoo',), ('ofoo',), ('pfoo',), ('qfoo',), ('rfoo',), ('sfoo',), ('tfoo',), ('ufoo',), ('vfoo',), ('wfoo',), ('xfoo',), ('yfoo',), ('zfoo',)]

cur.executescript('''
create table person(
firstname,
lastname,
age
);

create table book(
title,
author,
published
);

insert into book(title,author,published)
values(
'Dirk Gently''s Holistic Detective Agency',
'Douglas Adams',
1987
);
''')

<sqlite3.Cursor at 0xa36030>

cur.execute('select * from book')
cur.fetchall()

[("Dirk Gently's Holistic Detective Agencyfoo", 'Douglas Adamsfoo', 1987)]

cur.description

(('title', None, None, None, None, None, None),
('author', None, None, None, None, None, None),
('published', None, None, None, None, None, None))

class Point:
def __init__(self, x, y):
self.x, self.y = x, y
def __conform__(self,protocol):
if protocol is sqlite3.PrepareProtocol:
return '%f;%f'%(self.x,self.y)

con.text_factory=str

p=Point(4.0,-3.2)
cur.execute('select ?',(p,))
print(cur.fetchone()[0])

4.000000;-3.200000

class Point:
def __init__(self, x, y):
self.x, self.y = x, y

def __repr__(self):
return ("(%f;%f)" % (self.x, self.y))

def adapt_point(point):
return ('%f;%f'%(point.x,point.y)).encode('ascii')
sqlite3.register_adapter(Point,adapt_point)
p=Point(4.0,-3.2)
cur.execute('select ?',(p,))
print(cur.fetchone()[0])

b'4.000000;-3.200000'

import time
import datetime

def adapt_datetime(ts):
return time.mktime(ts.timetuple())

sqlite3.register_adapter(datetime.datetime,adapt_datetime)

now=datetime.datetime.now()
cur.execute('select ?',(now,))
print(cur.fetchone())

(1457593945.0,)

def convert_point(s):
x,y=list(map(float,s.split(b';')))
return Point(x,y)

sqlite3.register_converter('point',convert_point)

p=Point(4.0,-3.2)

#########################
# 1) Using declared types
con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES)
cur = con.cursor()
cur.execute("create table test(p point)")

cur.execute("insert into test(p) values (?)", (p,))
cur.execute("select p from test")
print("with declared types:", cur.fetchone()[0])
cur.close()
con.close()

with declared types: (4.000000;-3.200000)

#######################
# 1) Using column names
con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_COLNAMES)
cur = con.cursor()
cur.execute("create table test(p)")

cur.execute("insert into test(p) values (?)", (p,))
cur.execute('select p as "p [point]" from test')
print("with column names:", cur.fetchone()[0])
cur.close()
con.close()

with column names: (4.000000;-3.200000)

con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
cur = con.cursor()
cur.execute("create table test(d date, ts timestamp)")

<sqlite3.Cursor at 0xa23960>

today = datetime.date.today()
now = datetime.datetime.now()

cur.execute("insert into test(d, ts) values (?, ?)", (today, now))
print(today,now)

2016-03-10 2016-03-10 19:09:24.991262

cur.execute("select * from test")

---------------------------------------------------------------------------

ValueError                                Traceback (most recent call last)

<ipython-input-187-ff72650d6f80> in <module>()
----> 1 cur.execute("select * from test")

E:\Coding\Python3.4\lib\sqlite3\dbapi2.py in convert_timestamp(val)
65
66     def convert_timestamp(val):
---> 67         datepart, timepart = val.split(b" ")
68         year, month, day = map(int, datepart.split(b"-"))
69         timepart_full = timepart.split(b".")

ValueError: need more than 1 value to unpack

persons = [
("Hugo", "Boss"),
("Calvin", "Klein")
]

con = sqlite3.connect(":memory:")

# Create the table
con.execute("create table person(firstname, lastname)")

# Fill the table
con.executemany("insert into person(firstname, lastname) values (?, ?)", persons)

# Print the table contents
for row in con.execute("select firstname, lastname from person"):
print(row)

('Hugo', 'Boss')
('Calvin', 'Klein')

print("I just deleted", con.execute("delete from person").rowcount, "rows")

I just deleted 2 rows

con = sqlite3.connect(":memory:")
cur = con.cursor()

con.row_factory = sqlite3.Row
cur.execute("select 'John' as name, 42 as age")

<sqlite3.Cursor at 0xa239d0>

for row in cur:
assert row[0] == row["name"]
assert row["name"] == row["nAmE"]
assert row[1] == row["age"]
assert row[1] == row["AgE"]

---------------------------------------------------------------------------

TypeError                                 Traceback (most recent call last)

<ipython-input-195-fe5068d9accc> in <module>()
1 for row in cur:
----> 2     assert row[0] == row["name"]
3     assert row["name"] == row["nAmE"]
4     assert row[1] == row["age"]
5     assert row[1] == row["AgE"]

TypeError: tuple indices must be integers, not str

con = sqlite3.connect(":memory:")
con.execute("create table person (id integer primary key, firstname varchar unique)")

# Successful, con.commit() is called automatically afterwards
with con:
con.execute("insert into person(firstname) values (?)", ("Joe",))

# con.rollback() is called after the with block finishes with an exception, the
# exception is still raised and must be caught
try:
with con:
con.execute("insert into person(firstname) values (?)", ("Joe",))
except sqlite3.IntegrityError:
print("couldn't add Joe twice")

couldn't add Joe twice