traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

merged the sqlite-integration branch. 

This is based on pysqlite2.1.3, and provides a DB-API interface in
the standard library. You'll need sqlite 3.2.2 or later to build
this - if you have an earlier version, the C extension module will
not be built.
This commit is contained in:
Anthony Baxter 2006-04-01 00:57:31 +00:00
parent c17976e983
commit c51ee69b27
34 changed files with 6075 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
Lib/sqlite3/__init__.py Normal file
View File

@ -0,0 +1,24 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/__init__.py: the pysqlite2 package.
#
# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
from dbapi2 import *

84
Lib/sqlite3/dbapi2.py Normal file
View File

@ -0,0 +1,84 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/dbapi2.py: the DB-API 2.0 interface
#
# Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
import datetime
paramstyle = "qmark"
threadsafety = 1
apilevel = "2.0"
from _sqlite3 import *
import datetime, time
Date = datetime.date
Time = datetime.time
Timestamp = datetime.datetime
def DateFromTicks(ticks):
return apply(Date,time.localtime(ticks)[:3])
def TimeFromTicks(ticks):
return apply(Time,time.localtime(ticks)[3:6])
def TimestampFromTicks(ticks):
return apply(Timestamp,time.localtime(ticks)[:6])
_major, _minor, _micro = version.split(".")
version_info = (int(_major), int(_minor), _micro)
_major, _minor, _micro = sqlite_version.split(".")
sqlite_version_info = (int(_major), int(_minor), _micro)
Binary = buffer
def adapt_date(val):
return val.isoformat()
def adapt_datetime(val):
return val.isoformat(" ")
def convert_date(val):
return datetime.date(*map(int, val.split("-")))
def convert_timestamp(val):
datepart, timepart = val.split(" ")
year, month, day = map(int, datepart.split("-"))
timepart_full = timepart.split(".")
hours, minutes, seconds = map(int, timepart_full[0].split(":"))
if len(timepart_full) == 2:
microseconds = int(float("0." + timepart_full[1]) * 1000000)
else:
microseconds = 0
val = datetime.datetime(year, month, day, hours, minutes, seconds, microseconds)
return val
register_adapter(datetime.date, adapt_date)
register_adapter(datetime.datetime, adapt_datetime)
register_converter("date", convert_date)
register_converter("timestamp", convert_timestamp)

0
Lib/sqlite3/test/__init__.py Normal file
View File

726
Lib/sqlite3/test/dbapi.py Normal file
View File

@ -0,0 +1,726 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/dbapi.py: tests for DB-API compliance
#
# Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
import unittest
import threading
import sqlite3 as sqlite
class ModuleTests(unittest.TestCase):
def CheckAPILevel(self):
self.assertEqual(sqlite.apilevel, "2.0",
"apilevel is %s, should be 2.0" % sqlite.apilevel)
def CheckThreadSafety(self):
self.assertEqual(sqlite.threadsafety, 1,
"threadsafety is %d, should be 1" % sqlite.threadsafety)
def CheckParamStyle(self):
self.assertEqual(sqlite.paramstyle, "qmark",
"paramstyle is '%s', should be 'qmark'" %
sqlite.paramstyle)
def CheckWarning(self):
self.assert_(issubclass(sqlite.Warning, StandardError),
"Warning is not a subclass of StandardError")
def CheckError(self):
self.failUnless(issubclass(sqlite.Error, StandardError),
"Error is not a subclass of StandardError")
def CheckInterfaceError(self):
self.failUnless(issubclass(sqlite.InterfaceError, sqlite.Error),
"InterfaceError is not a subclass of Error")
def CheckDatabaseError(self):
self.failUnless(issubclass(sqlite.DatabaseError, sqlite.Error),
"DatabaseError is not a subclass of Error")
def CheckDataError(self):
self.failUnless(issubclass(sqlite.DataError, sqlite.DatabaseError),
"DataError is not a subclass of DatabaseError")
def CheckOperationalError(self):
self.failUnless(issubclass(sqlite.OperationalError, sqlite.DatabaseError),
"OperationalError is not a subclass of DatabaseError")
def CheckIntegrityError(self):
self.failUnless(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),
"IntegrityError is not a subclass of DatabaseError")
def CheckInternalError(self):
self.failUnless(issubclass(sqlite.InternalError, sqlite.DatabaseError),
"InternalError is not a subclass of DatabaseError")
def CheckProgrammingError(self):
self.failUnless(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),
"ProgrammingError is not a subclass of DatabaseError")
def CheckNotSupportedError(self):
self.failUnless(issubclass(sqlite.NotSupportedError,
sqlite.DatabaseError),
"NotSupportedError is not a subclass of DatabaseError")
class ConnectionTests(unittest.TestCase):
def setUp(self):
self.cx = sqlite.connect(":memory:")
cu = self.cx.cursor()
cu.execute("create table test(id integer primary key, name text)")
cu.execute("insert into test(name) values (?)", ("foo",))
def tearDown(self):
self.cx.close()
def CheckCommit(self):
self.cx.commit()
def CheckCommitAfterNoChanges(self):
"""
A commit should also work when no changes were made to the database.
"""
self.cx.commit()
self.cx.commit()
def CheckRollback(self):
self.cx.rollback()
def CheckRollbackAfterNoChanges(self):
"""
A rollback should also work when no changes were made to the database.
"""
self.cx.rollback()
self.cx.rollback()
def CheckCursor(self):
cu = self.cx.cursor()
def CheckFailedOpen(self):
YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db"
try:
con = sqlite.connect(YOU_CANNOT_OPEN_THIS)
except sqlite.OperationalError:
return
self.fail("should have raised an OperationalError")
def CheckClose(self):
self.cx.close()
def CheckExceptions(self):
# Optional DB-API extension.
self.failUnlessEqual(self.cx.Warning, sqlite.Warning)
self.failUnlessEqual(self.cx.Error, sqlite.Error)
self.failUnlessEqual(self.cx.InterfaceError, sqlite.InterfaceError)
self.failUnlessEqual(self.cx.DatabaseError, sqlite.DatabaseError)
self.failUnlessEqual(self.cx.DataError, sqlite.DataError)
self.failUnlessEqual(self.cx.OperationalError, sqlite.OperationalError)
self.failUnlessEqual(self.cx.IntegrityError, sqlite.IntegrityError)
self.failUnlessEqual(self.cx.InternalError, sqlite.InternalError)
self.failUnlessEqual(self.cx.ProgrammingError, sqlite.ProgrammingError)
self.failUnlessEqual(self.cx.NotSupportedError, sqlite.NotSupportedError)
class CursorTests(unittest.TestCase):
def setUp(self):
self.cx = sqlite.connect(":memory:")
self.cu = self.cx.cursor()
self.cu.execute("create table test(id integer primary key, name text, income number)")
self.cu.execute("insert into test(name) values (?)", ("foo",))
def tearDown(self):
self.cu.close()
self.cx.close()
def CheckExecuteNoArgs(self):
self.cu.execute("delete from test")
def CheckExecuteIllegalSql(self):
try:
self.cu.execute("select asdf")
self.fail("should have raised an OperationalError")
except sqlite.OperationalError:
return
except:
self.fail("raised wrong exception")
def CheckExecuteTooMuchSql(self):
try:
self.cu.execute("select 5+4; select 4+5")
self.fail("should have raised a Warning")
except sqlite.Warning:
return
except:
self.fail("raised wrong exception")
def CheckExecuteTooMuchSql2(self):
self.cu.execute("select 5+4; -- foo bar")
def CheckExecuteTooMuchSql3(self):
self.cu.execute("""
select 5+4;
/*
foo
*/
""")
def CheckExecuteWrongSqlArg(self):
try:
self.cu.execute(42)
self.fail("should have raised a ValueError")
except ValueError:
return
except:
self.fail("raised wrong exception.")
def CheckExecuteArgInt(self):
self.cu.execute("insert into test(id) values (?)", (42,))
def CheckExecuteArgFloat(self):
self.cu.execute("insert into test(income) values (?)", (2500.32,))
def CheckExecuteArgString(self):
self.cu.execute("insert into test(name) values (?)", ("Hugo",))
def CheckExecuteWrongNoOfArgs1(self):
# too many parameters
try:
self.cu.execute("insert into test(id) values (?)", (17, "Egon"))
self.fail("should have raised ProgrammingError")
except sqlite.ProgrammingError:
pass
def CheckExecuteWrongNoOfArgs2(self):
# too little parameters
try:
self.cu.execute("insert into test(id) values (?)")
self.fail("should have raised ProgrammingError")
except sqlite.ProgrammingError:
pass
def CheckExecuteWrongNoOfArgs3(self):
# no parameters, parameters are needed
try:
self.cu.execute("insert into test(id) values (?)")
self.fail("should have raised ProgrammingError")
except sqlite.ProgrammingError:
pass
def CheckExecuteDictMapping(self):
self.cu.execute("insert into test(name) values ('foo')")
self.cu.execute("select name from test where name=:name", {"name": "foo"})
row = self.cu.fetchone()
self.failUnlessEqual(row[0], "foo")
def CheckExecuteDictMappingTooLittleArgs(self):
self.cu.execute("insert into test(name) values ('foo')")
try:
self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"})
self.fail("should have raised ProgrammingError")
except sqlite.ProgrammingError:
pass
def CheckExecuteDictMappingNoArgs(self):
self.cu.execute("insert into test(name) values ('foo')")
try:
self.cu.execute("select name from test where name=:name")
self.fail("should have raised ProgrammingError")
except sqlite.ProgrammingError:
pass
def CheckExecuteDictMappingUnnamed(self):
self.cu.execute("insert into test(name) values ('foo')")
try:
self.cu.execute("select name from test where name=?", {"name": "foo"})
self.fail("should have raised ProgrammingError")
except sqlite.ProgrammingError:
pass
def CheckClose(self):
self.cu.close()
def CheckRowcountExecute(self):
self.cu.execute("delete from test")
self.cu.execute("insert into test(name) values ('foo')")
self.cu.execute("insert into test(name) values ('foo')")
self.cu.execute("update test set name='bar'")
self.failUnlessEqual(self.cu.rowcount, 2)
def CheckRowcountExecutemany(self):
self.cu.execute("delete from test")
self.cu.executemany("insert into test(name) values (?)", [(1,), (2,), (3,)])
self.failUnlessEqual(self.cu.rowcount, 3)
# Checks for executemany:
# Sequences are required by the DB-API, iterators
# enhancements in pysqlite.
def CheckExecuteManySequence(self):
self.cu.executemany("insert into test(income) values (?)", [(x,) for x in range(100, 110)])
def CheckExecuteManyIterator(self):
class MyIter:
def __init__(self):
self.value = 5
def next(self):
if self.value == 10:
raise StopIteration
else:
self.value += 1
return (self.value,)
self.cu.executemany("insert into test(income) values (?)", MyIter())
def CheckExecuteManyGenerator(self):
def mygen():
for i in range(5):
yield (i,)
self.cu.executemany("insert into test(income) values (?)", mygen())
def CheckExecuteManyWrongSqlArg(self):
try:
self.cu.executemany(42, [(3,)])
self.fail("should have raised a ValueError")
except ValueError:
return
except:
self.fail("raised wrong exception.")
def CheckExecuteManySelect(self):
try:
self.cu.executemany("select ?", [(3,)])
self.fail("should have raised a ProgrammingError")
except sqlite.ProgrammingError:
return
except:
self.fail("raised wrong exception.")
def CheckExecuteManyNotIterable(self):
try:
self.cu.executemany("insert into test(income) values (?)", 42)
self.fail("should have raised a TypeError")
except TypeError:
return
except Exception, e:
print "raised", e.__class__
self.fail("raised wrong exception.")
def CheckFetchIter(self):
# Optional DB-API extension.
self.cu.execute("delete from test")
self.cu.execute("insert into test(id) values (?)", (5,))
self.cu.execute("insert into test(id) values (?)", (6,))
self.cu.execute("select id from test order by id")
lst = []
for row in self.cu:
lst.append(row[0])
self.failUnlessEqual(lst[0], 5)
self.failUnlessEqual(lst[1], 6)
def CheckFetchone(self):
self.cu.execute("select name from test")
row = self.cu.fetchone()
self.failUnlessEqual(row[0], "foo")
row = self.cu.fetchone()
self.failUnlessEqual(row, None)
def CheckFetchoneNoStatement(self):
cur = self.cx.cursor()
row = cur.fetchone()
self.failUnlessEqual(row, None)
def CheckArraySize(self):
# must default ot 1
self.failUnlessEqual(self.cu.arraysize, 1)
# now set to 2
self.cu.arraysize = 2
# now make the query return 3 rows
self.cu.execute("delete from test")
self.cu.execute("insert into test(name) values ('A')")
self.cu.execute("insert into test(name) values ('B')")
self.cu.execute("insert into test(name) values ('C')")
self.cu.execute("select name from test")
res = self.cu.fetchmany()
self.failUnlessEqual(len(res), 2)
def CheckFetchmany(self):
self.cu.execute("select name from test")
res = self.cu.fetchmany(100)
self.failUnlessEqual(len(res), 1)
res = self.cu.fetchmany(100)
self.failUnlessEqual(res, [])
def CheckFetchall(self):
self.cu.execute("select name from test")
res = self.cu.fetchall()
self.failUnlessEqual(len(res), 1)
res = self.cu.fetchall()
self.failUnlessEqual(res, [])
def CheckSetinputsizes(self):
self.cu.setinputsizes([3, 4, 5])
def CheckSetoutputsize(self):
self.cu.setoutputsize(5, 0)
def CheckSetoutputsizeNoColumn(self):
self.cu.setoutputsize(42)
def CheckCursorConnection(self):
# Optional DB-API extension.
self.failUnlessEqual(self.cu.connection, self.cx)
def CheckWrongCursorCallable(self):
try:
def f(): pass
cur = self.cx.cursor(f)
self.fail("should have raised a TypeError")
except TypeError:
return
self.fail("should have raised a ValueError")
def CheckCursorWrongClass(self):
class Foo: pass
foo = Foo()
try:
cur = sqlite.Cursor(foo)
self.fail("should have raised a ValueError")
except TypeError:
pass
class ThreadTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
self.cur = self.con.cursor()
self.cur.execute("create table test(id integer primary key, name text, bin binary, ratio number, ts timestamp)")
def tearDown(self):
self.cur.close()
self.con.close()
def CheckConCursor(self):
def run(con, errors):
try:
cur = con.cursor()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckConCommit(self):
def run(con, errors):
try:
con.commit()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckConRollback(self):
def run(con, errors):
try:
con.rollback()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckConClose(self):
def run(con, errors):
try:
con.close()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckCurImplicitBegin(self):
def run(cur, errors):
try:
cur.execute("insert into test(name) values ('a')")
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckCurClose(self):
def run(cur, errors):
try:
cur.close()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckCurExecute(self):
def run(cur, errors):
try:
cur.execute("select name from test")
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
self.cur.execute("insert into test(name) values ('a')")
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
def CheckCurIterNext(self):
def run(cur, errors):
try:
row = cur.fetchone()
errors.append("did not raise ProgrammingError")
return
except sqlite.ProgrammingError:
return
except:
errors.append("raised wrong exception")
errors = []
self.cur.execute("insert into test(name) values ('a')")
self.cur.execute("select name from test")
t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
t.start()
t.join()
if len(errors) > 0:
self.fail("\n".join(errors))
class ConstructorTests(unittest.TestCase):
def CheckDate(self):
d = sqlite.Date(2004, 10, 28)
def CheckTime(self):
t = sqlite.Time(12, 39, 35)
def CheckTimestamp(self):
ts = sqlite.Timestamp(2004, 10, 28, 12, 39, 35)
def CheckDateFromTicks(self):
d = sqlite.DateFromTicks(42)
def CheckTimeFromTicks(self):
t = sqlite.TimeFromTicks(42)
def CheckTimestampFromTicks(self):
ts = sqlite.TimestampFromTicks(42)
def CheckBinary(self):
b = sqlite.Binary(chr(0) + "'")
class ExtensionTests(unittest.TestCase):
def CheckScriptStringSql(self):
con = sqlite.connect(":memory:")
cur = con.cursor()
cur.executescript("""
-- bla bla
/* a stupid comment */
create table a(i);
insert into a(i) values (5);
""")
cur.execute("select i from a")
res = cur.fetchone()[0]
self.failUnlessEqual(res, 5)
def CheckScriptStringUnicode(self):
con = sqlite.connect(":memory:")
cur = con.cursor()
cur.executescript(u"""
create table a(i);
insert into a(i) values (5);
select i from a;
delete from a;
insert into a(i) values (6);
""")
cur.execute("select i from a")
res = cur.fetchone()[0]
self.failUnlessEqual(res, 6)
def CheckScriptErrorIncomplete(self):
con = sqlite.connect(":memory:")
cur = con.cursor()
raised = False
try:
cur.executescript("create table test(sadfsadfdsa")
except sqlite.ProgrammingError:
raised = True
self.failUnlessEqual(raised, True, "should have raised an exception")
def CheckScriptErrorNormal(self):
con = sqlite.connect(":memory:")
cur = con.cursor()
raised = False
try:
cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")
except sqlite.OperationalError:
raised = True
self.failUnlessEqual(raised, True, "should have raised an exception")
def CheckConnectionExecute(self):
con = sqlite.connect(":memory:")
result = con.execute("select 5").fetchone()[0]
self.failUnlessEqual(result, 5, "Basic test of Connection.execute")
def CheckConnectionExecutemany(self):
con = sqlite.connect(":memory:")
con.execute("create table test(foo)")
con.executemany("insert into test(foo) values (?)", [(3,), (4,)])
result = con.execute("select foo from test order by foo").fetchall()
self.failUnlessEqual(result[0][0], 3, "Basic test of Connection.executemany")
self.failUnlessEqual(result[1][0], 4, "Basic test of Connection.executemany")
def CheckConnectionExecutescript(self):
con = sqlite.connect(":memory:")
con.executescript("create table test(foo); insert into test(foo) values (5);")
result = con.execute("select foo from test").fetchone()[0]
self.failUnlessEqual(result, 5, "Basic test of Connection.executescript")
class ClosedTests(unittest.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
def CheckClosedConCursor(self):
con = sqlite.connect(":memory:")
con.close()
try:
cur = con.cursor()
self.fail("Should have raised a ProgrammingError")
except sqlite.ProgrammingError:
pass
except:
self.fail("Should have raised a ProgrammingError")
def CheckClosedConCommit(self):
con = sqlite.connect(":memory:")
con.close()
try:
con.commit()
self.fail("Should have raised a ProgrammingError")
except sqlite.ProgrammingError:
pass
except:
self.fail("Should have raised a ProgrammingError")
def CheckClosedConRollback(self):
con = sqlite.connect(":memory:")
con.close()
try:
con.rollback()
self.fail("Should have raised a ProgrammingError")
except sqlite.ProgrammingError:
pass
except:
self.fail("Should have raised a ProgrammingError")
def CheckClosedCurExecute(self):
con = sqlite.connect(":memory:")
cur = con.cursor()
con.close()
try:
cur.execute("select 4")
self.fail("Should have raised a ProgrammingError")
except sqlite.ProgrammingError:
pass
except:
self.fail("Should have raised a ProgrammingError")
def suite():
module_suite = unittest.makeSuite(ModuleTests, "Check")
connection_suite = unittest.makeSuite(ConnectionTests, "Check")
cursor_suite = unittest.makeSuite(CursorTests, "Check")
thread_suite = unittest.makeSuite(ThreadTests, "Check")
constructor_suite = unittest.makeSuite(ConstructorTests, "Check")
ext_suite = unittest.makeSuite(ExtensionTests, "Check")
closed_suite = unittest.makeSuite(ClosedTests, "Check")
return unittest.TestSuite((module_suite, connection_suite, cursor_suite, thread_suite, constructor_suite, ext_suite, closed_suite))
def test():
runner = unittest.TextTestRunner()
runner.run(suite())
if __name__ == "__main__":
test()

164
Lib/sqlite3/test/factory.py Normal file
View File

@ -0,0 +1,164 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/factory.py: tests for the various factories in pysqlite
#
# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
import unittest
import sqlite3 as sqlite
class MyConnection(sqlite.Connection):
def __init__(self, *args, **kwargs):
sqlite.Connection.__init__(self, *args, **kwargs)
def dict_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
d[col[0]] = row[idx]
return d
class MyCursor(sqlite.Cursor):
def __init__(self, *args, **kwargs):
sqlite.Cursor.__init__(self, *args, **kwargs)
self.row_factory = dict_factory
class ConnectionFactoryTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:", factory=MyConnection)
def tearDown(self):
self.con.close()
def CheckIsInstance(self):
self.failUnless(isinstance(self.con,
MyConnection),
"connection is not instance of MyConnection")
class CursorFactoryTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
def tearDown(self):
self.con.close()
def CheckIsInstance(self):
cur = self.con.cursor(factory=MyCursor)
self.failUnless(isinstance(cur,
MyCursor),
"cursor is not instance of MyCursor")
class RowFactoryTestsBackwardsCompat(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
def CheckIsProducedByFactory(self):
cur = self.con.cursor(factory=MyCursor)
cur.execute("select 4+5 as foo")
row = cur.fetchone()
self.failUnless(isinstance(row,
dict),
"row is not instance of dict")
cur.close()
def tearDown(self):
self.con.close()
class RowFactoryTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
def CheckCustomFactory(self):
self.con.row_factory = lambda cur, row: list(row)
row = self.con.execute("select 1, 2").fetchone()
self.failUnless(isinstance(row,
list),
"row is not instance of list")
def CheckSqliteRow(self):
self.con.row_factory = sqlite.Row
row = self.con.execute("select 1 as a, 2 as b").fetchone()
self.failUnless(isinstance(row,
sqlite.Row),
"row is not instance of sqlite.Row")
col1, col2 = row["a"], row["b"]
self.failUnless(col1 == 1, "by name: wrong result for column 'a'")
self.failUnless(col2 == 2, "by name: wrong result for column 'a'")
col1, col2 = row["A"], row["B"]
self.failUnless(col1 == 1, "by name: wrong result for column 'A'")
self.failUnless(col2 == 2, "by name: wrong result for column 'B'")
col1, col2 = row[0], row[1]
self.failUnless(col1 == 1, "by index: wrong result for column 0")
self.failUnless(col2 == 2, "by index: wrong result for column 1")
def tearDown(self):
self.con.close()
class TextFactoryTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
def CheckUnicode(self):
austria = unicode("Österreich", "latin1")
row = self.con.execute("select ?", (austria,)).fetchone()
self.failUnless(type(row[0]) == unicode, "type of row[0] must be unicode")
def CheckString(self):
self.con.text_factory = str
austria = unicode("Österreich", "latin1")
row = self.con.execute("select ?", (austria,)).fetchone()
self.failUnless(type(row[0]) == str, "type of row[0] must be str")
self.failUnless(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8")
def CheckCustom(self):
self.con.text_factory = lambda x: unicode(x, "utf-8", "ignore")
austria = unicode("Österreich", "latin1")
row = self.con.execute("select ?", (austria.encode("latin1"),)).fetchone()
self.failUnless(type(row[0]) == unicode, "type of row[0] must be unicode")
self.failUnless(row[0].endswith(u"reich"), "column must contain original data")
def CheckOptimizedUnicode(self):
self.con.text_factory = sqlite.OptimizedUnicode
austria = unicode("Österreich", "latin1")
germany = unicode("Deutchland")
a_row = self.con.execute("select ?", (austria,)).fetchone()
d_row = self.con.execute("select ?", (germany,)).fetchone()
self.failUnless(type(a_row[0]) == unicode, "type of non-ASCII row must be unicode")
self.failUnless(type(d_row[0]) == str, "type of ASCII-only row must be str")
def tearDown(self):
self.con.close()
def suite():
connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check")
cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check")
row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check")
row_suite = unittest.makeSuite(RowFactoryTests, "Check")
text_suite = unittest.makeSuite(TextFactoryTests, "Check")
return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite))
def test():
runner = unittest.TextTestRunner()
runner.run(suite())
if __name__ == "__main__":
test()

154
Lib/sqlite3/test/transactions.py Normal file
View File

@ -0,0 +1,154 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/transactions.py: tests transactions
#
# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
import os, unittest
import sqlite3 as sqlite
def get_db_path():
return "testdb"
class TransactionTests(unittest.TestCase):
def setUp(self):
try:
os.remove(get_db_path())
except:
pass
self.con1 = sqlite.connect(get_db_path(), timeout=0.1)
self.cur1 = self.con1.cursor()
self.con2 = sqlite.connect(get_db_path(), timeout=0.1)
self.cur2 = self.con2.cursor()
def tearDown(self):
self.cur1.close()
self.con1.close()
self.cur2.close()
self.con2.close()
def CheckDMLdoesAutoCommitBefore(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.cur1.execute("create table test2(j)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 1)
def CheckInsertStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 0)
def CheckUpdateStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.con1.commit()
self.cur1.execute("update test set i=6")
self.cur2.execute("select i from test")
res = self.cur2.fetchone()[0]
self.failUnlessEqual(res, 5)
def CheckDeleteStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.con1.commit()
self.cur1.execute("delete from test")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 1)
def CheckReplaceStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.con1.commit()
self.cur1.execute("replace into test(i) values (6)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 1)
self.failUnlessEqual(res[0][0], 5)
def CheckToggleAutoCommit(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.con1.isolation_level = None
self.failUnlessEqual(self.con1.isolation_level, None)
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 1)
self.con1.isolation_level = "DEFERRED"
self.failUnlessEqual(self.con1.isolation_level , "DEFERRED")
self.cur1.execute("insert into test(i) values (5)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
self.failUnlessEqual(len(res), 1)
def CheckRaiseTimeout(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
try:
self.cur2.execute("insert into test(i) values (5)")
self.fail("should have raised an OperationalError")
except sqlite.OperationalError:
pass
except:
self.fail("should have raised an OperationalError")
class SpecialCommandTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
self.cur = self.con.cursor()
def CheckVacuum(self):
self.cur.execute("create table test(i)")
self.cur.execute("insert into test(i) values (5)")
self.cur.execute("vacuum")
def CheckDropTable(self):
self.cur.execute("create table test(i)")
self.cur.execute("insert into test(i) values (5)")
self.cur.execute("drop table test")
def CheckPragma(self):
self.cur.execute("create table test(i)")
self.cur.execute("insert into test(i) values (5)")
self.cur.execute("pragma count_changes=1")
def tearDown(self):
self.cur.close()
self.con.close()
def suite():
default_suite = unittest.makeSuite(TransactionTests, "Check")
special_command_suite = unittest.makeSuite(SpecialCommandTests, "Check")
return unittest.TestSuite((default_suite, special_command_suite))
def test():
runner = unittest.TextTestRunner()
runner.run(suite())
if __name__ == "__main__":
test()

339
Lib/sqlite3/test/types.py Normal file
View File

@ -0,0 +1,339 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/types.py: tests for type conversion and detection
#
# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
import datetime
import unittest
import sqlite3 as sqlite
class SqliteTypeTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
self.cur = self.con.cursor()
self.cur.execute("create table test(i integer, s varchar, f number, b blob)")
def tearDown(self):
self.cur.close()
self.con.close()
def CheckString(self):
self.cur.execute("insert into test(s) values (?)", (u"Österreich",))
self.cur.execute("select s from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], u"Österreich")
def CheckSmallInt(self):
self.cur.execute("insert into test(i) values (?)", (42,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], 42)
def CheckLargeInt(self):
num = 2**40
self.cur.execute("insert into test(i) values (?)", (num,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], num)
def CheckFloat(self):
val = 3.14
self.cur.execute("insert into test(f) values (?)", (val,))
self.cur.execute("select f from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], val)
def CheckBlob(self):
val = buffer("Guglhupf")
self.cur.execute("insert into test(b) values (?)", (val,))
self.cur.execute("select b from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], val)
def CheckUnicodeExecute(self):
self.cur.execute(u"select 'Österreich'")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], u"Österreich")
class DeclTypesTests(unittest.TestCase):
class Foo:
def __init__(self, _val):
self.val = _val
def __cmp__(self, other):
if not isinstance(other, DeclTypesTests.Foo):
raise ValueError
if self.val == other.val:
return 0
else:
return 1
def __conform__(self, protocol):
if protocol is sqlite.PrepareProtocol:
return self.val
else:
return None
def __str__(self):
return "<%s>" % self.val
def setUp(self):
self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
self.cur = self.con.cursor()
self.cur.execute("create table test(i int, s str, f float, b bool, u unicode, foo foo, bin blob)")
# override float, make them always return the same number
sqlite.converters["float"] = lambda x: 47.2
# and implement two custom ones
sqlite.converters["bool"] = lambda x: bool(int(x))
sqlite.converters["foo"] = DeclTypesTests.Foo
def tearDown(self):
del sqlite.converters["float"]
del sqlite.converters["bool"]
del sqlite.converters["foo"]
self.cur.close()
self.con.close()
def CheckString(self):
# default
self.cur.execute("insert into test(s) values (?)", ("foo",))
self.cur.execute("select s from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], "foo")
def CheckSmallInt(self):
# default
self.cur.execute("insert into test(i) values (?)", (42,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], 42)
def CheckLargeInt(self):
# default
num = 2**40
self.cur.execute("insert into test(i) values (?)", (num,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], num)
def CheckFloat(self):
# custom
val = 3.14
self.cur.execute("insert into test(f) values (?)", (val,))
self.cur.execute("select f from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], 47.2)
def CheckBool(self):
# custom
self.cur.execute("insert into test(b) values (?)", (False,))
self.cur.execute("select b from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], False)
self.cur.execute("delete from test")
self.cur.execute("insert into test(b) values (?)", (True,))
self.cur.execute("select b from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], True)
def CheckUnicode(self):
# default
val = u"\xd6sterreich"
self.cur.execute("insert into test(u) values (?)", (val,))
self.cur.execute("select u from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], val)
def CheckFoo(self):
val = DeclTypesTests.Foo("bla")
self.cur.execute("insert into test(foo) values (?)", (val,))
self.cur.execute("select foo from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], val)
def CheckUnsupportedSeq(self):
class Bar: pass
val = Bar()
try:
self.cur.execute("insert into test(f) values (?)", (val,))
self.fail("should have raised an InterfaceError")
except sqlite.InterfaceError:
pass
except:
self.fail("should have raised an InterfaceError")
def CheckUnsupportedDict(self):
class Bar: pass
val = Bar()
try:
self.cur.execute("insert into test(f) values (:val)", {"val": val})
self.fail("should have raised an InterfaceError")
except sqlite.InterfaceError:
pass
except:
self.fail("should have raised an InterfaceError")
def CheckBlob(self):
# default
val = buffer("Guglhupf")
self.cur.execute("insert into test(bin) values (?)", (val,))
self.cur.execute("select bin from test")
row = self.cur.fetchone()
self.failUnlessEqual(row[0], val)
class ColNamesTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES|sqlite.PARSE_DECLTYPES)
self.cur = self.con.cursor()
self.cur.execute("create table test(x foo)")
sqlite.converters["foo"] = lambda x: "[%s]" % x
sqlite.converters["bar"] = lambda x: "<%s>" % x
sqlite.converters["exc"] = lambda x: 5/0
def tearDown(self):
del sqlite.converters["foo"]
del sqlite.converters["bar"]
del sqlite.converters["exc"]
self.cur.close()
self.con.close()
def CheckDeclType(self):
self.cur.execute("insert into test(x) values (?)", ("xxx",))
self.cur.execute("select x from test")
val = self.cur.fetchone()[0]
self.failUnlessEqual(val, "[xxx]")
def CheckNone(self):
self.cur.execute("insert into test(x) values (?)", (None,))
self.cur.execute("select x from test")
val = self.cur.fetchone()[0]
self.failUnlessEqual(val, None)
def CheckExc(self):
# Exceptions in type converters result in returned Nones
self.cur.execute('select 5 as "x [exc]"')
val = self.cur.fetchone()[0]
self.failUnlessEqual(val, None)
def CheckColName(self):
self.cur.execute("insert into test(x) values (?)", ("xxx",))
self.cur.execute('select x as "x [bar]" from test')
val = self.cur.fetchone()[0]
self.failUnlessEqual(val, "<xxx>")
# Check if the stripping of colnames works. Everything after the first
# whitespace should be stripped.
self.failUnlessEqual(self.cur.description[0][0], "x")
def CheckCursorDescriptionNoRow(self):
"""
cursor.description should at least provide the column name(s), even if
no row returned.
"""
self.cur.execute("select * from test where 0 = 1")
self.assert_(self.cur.description[0][0] == "x")
class ObjectAdaptationTests(unittest.TestCase):
def cast(obj):
return float(obj)
cast = staticmethod(cast)
def setUp(self):
self.con = sqlite.connect(":memory:")
try:
del sqlite.adapters[int]
except:
pass
sqlite.register_adapter(int, ObjectAdaptationTests.cast)
self.cur = self.con.cursor()
def tearDown(self):
del sqlite.adapters[(int, sqlite.PrepareProtocol)]
self.cur.close()
self.con.close()
def CheckCasterIsUsed(self):
self.cur.execute("select ?", (4,))
val = self.cur.fetchone()[0]
self.failUnlessEqual(type(val), float)
class DateTimeTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
self.cur = self.con.cursor()
self.cur.execute("create table test(d date, ts timestamp)")
def tearDown(self):
self.cur.close()
self.con.close()
def CheckSqliteDate(self):
d = sqlite.Date(2004, 2, 14)
self.cur.execute("insert into test(d) values (?)", (d,))
self.cur.execute("select d from test")
d2 = self.cur.fetchone()[0]
self.failUnlessEqual(d, d2)
def CheckSqliteTimestamp(self):
ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0)
self.cur.execute("insert into test(ts) values (?)", (ts,))
self.cur.execute("select ts from test")
ts2 = self.cur.fetchone()[0]
self.failUnlessEqual(ts, ts2)
def CheckSqlTimestamp(self):
# The date functions are only available in SQLite version 3.1 or later
if sqlite.sqlite_version_info < (3, 1):
return
# SQLite's current_timestamp uses UTC time, while datetime.datetime.now() uses local time.
now = datetime.datetime.now()
self.cur.execute("insert into test(ts) values (current_timestamp)")
self.cur.execute("select ts from test")
ts = self.cur.fetchone()[0]
self.failUnlessEqual(type(ts), datetime.datetime)
self.failUnlessEqual(ts.year, now.year)
def CheckDateTimeSubSeconds(self):
ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 500000)
self.cur.execute("insert into test(ts) values (?)", (ts,))
self.cur.execute("select ts from test")
ts2 = self.cur.fetchone()[0]
self.failUnlessEqual(ts, ts2)
def suite():
sqlite_type_suite = unittest.makeSuite(SqliteTypeTests, "Check")
decltypes_type_suite = unittest.makeSuite(DeclTypesTests, "Check")
colnames_type_suite = unittest.makeSuite(ColNamesTests, "Check")
adaptation_suite = unittest.makeSuite(ObjectAdaptationTests, "Check")
date_suite = unittest.makeSuite(DateTimeTests, "Check")
return unittest.TestSuite((sqlite_type_suite, decltypes_type_suite, colnames_type_suite, adaptation_suite, date_suite))
def test():
runner = unittest.TextTestRunner()
runner.run(suite())
if __name__ == "__main__":
test()

330
Lib/sqlite3/test/userfunctions.py Normal file
View File

@ -0,0 +1,330 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/userfunctions.py: tests for user-defined functions and
# aggregates.
#
# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
import unittest
import sqlite3 as sqlite
def func_returntext():
return "foo"
def func_returnunicode():
return u"bar"
def func_returnint():
return 42
def func_returnfloat():
return 3.14
def func_returnnull():
return None
def func_returnblob():
return buffer("blob")
def func_raiseexception():
5/0
def func_isstring(v):
return type(v) is unicode
def func_isint(v):
return type(v) is int
def func_isfloat(v):
return type(v) is float
def func_isnone(v):
return type(v) is type(None)
def func_isblob(v):
return type(v) is buffer
class AggrNoStep:
def __init__(self):
pass
class AggrNoFinalize:
def __init__(self):
pass
def step(self, x):
pass
class AggrExceptionInInit:
def __init__(self):
5/0
def step(self, x):
pass
def finalize(self):
pass
class AggrExceptionInStep:
def __init__(self):
pass
def step(self, x):
5/0
def finalize(self):
return 42
class AggrExceptionInFinalize:
def __init__(self):
pass
def step(self, x):
pass
def finalize(self):
5/0
class AggrCheckType:
def __init__(self):
self.val = None
def step(self, whichType, val):
theType = {"str": unicode, "int": int, "float": float, "None": type(None), "blob": buffer}
self.val = int(theType[whichType] is type(val))
def finalize(self):
return self.val
class AggrSum:
def __init__(self):
self.val = 0.0
def step(self, val):
self.val += val
def finalize(self):
return self.val
class FunctionTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
self.con.create_function("returntext", 0, func_returntext)
self.con.create_function("returnunicode", 0, func_returnunicode)
self.con.create_function("returnint", 0, func_returnint)
self.con.create_function("returnfloat", 0, func_returnfloat)
self.con.create_function("returnnull", 0, func_returnnull)
self.con.create_function("returnblob", 0, func_returnblob)
self.con.create_function("raiseexception", 0, func_raiseexception)
self.con.create_function("isstring", 1, func_isstring)
self.con.create_function("isint", 1, func_isint)
self.con.create_function("isfloat", 1, func_isfloat)
self.con.create_function("isnone", 1, func_isnone)
self.con.create_function("isblob", 1, func_isblob)
def tearDown(self):
self.con.close()
def CheckFuncRefCount(self):
def getfunc():
def f():
return val
return f
self.con.create_function("reftest", 0, getfunc())
cur = self.con.cursor()
cur.execute("select reftest()")
def CheckFuncReturnText(self):
cur = self.con.cursor()
cur.execute("select returntext()")
val = cur.fetchone()[0]
self.failUnlessEqual(type(val), unicode)
self.failUnlessEqual(val, "foo")
def CheckFuncReturnUnicode(self):
cur = self.con.cursor()
cur.execute("select returnunicode()")
val = cur.fetchone()[0]
self.failUnlessEqual(type(val), unicode)
self.failUnlessEqual(val, u"bar")
def CheckFuncReturnInt(self):
cur = self.con.cursor()
cur.execute("select returnint()")
val = cur.fetchone()[0]
self.failUnlessEqual(type(val), int)
self.failUnlessEqual(val, 42)
def CheckFuncReturnFloat(self):
cur = self.con.cursor()
cur.execute("select returnfloat()")
val = cur.fetchone()[0]
self.failUnlessEqual(type(val), float)
if val < 3.139 or val > 3.141:
self.fail("wrong value")
def CheckFuncReturnNull(self):
cur = self.con.cursor()
cur.execute("select returnnull()")
val = cur.fetchone()[0]
self.failUnlessEqual(type(val), type(None))
self.failUnlessEqual(val, None)
def CheckFuncReturnBlob(self):
cur = self.con.cursor()
cur.execute("select returnblob()")
val = cur.fetchone()[0]
self.failUnlessEqual(type(val), buffer)
self.failUnlessEqual(val, buffer("blob"))
def CheckFuncException(self):
cur = self.con.cursor()
cur.execute("select raiseexception()")
val = cur.fetchone()[0]
self.failUnlessEqual(val, None)
def CheckParamString(self):
cur = self.con.cursor()
cur.execute("select isstring(?)", ("foo",))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckParamInt(self):
cur = self.con.cursor()
cur.execute("select isint(?)", (42,))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckParamFloat(self):
cur = self.con.cursor()
cur.execute("select isfloat(?)", (3.14,))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckParamNone(self):
cur = self.con.cursor()
cur.execute("select isnone(?)", (None,))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckParamBlob(self):
cur = self.con.cursor()
cur.execute("select isblob(?)", (buffer("blob"),))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
class AggregateTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
cur = self.con.cursor()
cur.execute("""
create table test(
t text,
i integer,
f float,
n,
b blob
)
""")
cur.execute("insert into test(t, i, f, n, b) values (?, ?, ?, ?, ?)",
("foo", 5, 3.14, None, buffer("blob"),))
self.con.create_aggregate("nostep", 1, AggrNoStep)
self.con.create_aggregate("nofinalize", 1, AggrNoFinalize)
self.con.create_aggregate("excInit", 1, AggrExceptionInInit)
self.con.create_aggregate("excStep", 1, AggrExceptionInStep)
self.con.create_aggregate("excFinalize", 1, AggrExceptionInFinalize)
self.con.create_aggregate("checkType", 2, AggrCheckType)
self.con.create_aggregate("mysum", 1, AggrSum)
def tearDown(self):
#self.cur.close()
#self.con.close()
pass
def CheckAggrNoStep(self):
cur = self.con.cursor()
cur.execute("select nostep(t) from test")
def CheckAggrNoFinalize(self):
cur = self.con.cursor()
cur.execute("select nofinalize(t) from test")
val = cur.fetchone()[0]
self.failUnlessEqual(val, None)
def CheckAggrExceptionInInit(self):
cur = self.con.cursor()
cur.execute("select excInit(t) from test")
val = cur.fetchone()[0]
self.failUnlessEqual(val, None)
def CheckAggrExceptionInStep(self):
cur = self.con.cursor()
cur.execute("select excStep(t) from test")
val = cur.fetchone()[0]
self.failUnlessEqual(val, 42)
def CheckAggrExceptionInFinalize(self):
cur = self.con.cursor()
cur.execute("select excFinalize(t) from test")
val = cur.fetchone()[0]
self.failUnlessEqual(val, None)
def CheckAggrCheckParamStr(self):
cur = self.con.cursor()
cur.execute("select checkType('str', ?)", ("foo",))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckAggrCheckParamInt(self):
cur = self.con.cursor()
cur.execute("select checkType('int', ?)", (42,))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckAggrCheckParamFloat(self):
cur = self.con.cursor()
cur.execute("select checkType('float', ?)", (3.14,))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckAggrCheckParamNone(self):
cur = self.con.cursor()
cur.execute("select checkType('None', ?)", (None,))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckAggrCheckParamBlob(self):
cur = self.con.cursor()
cur.execute("select checkType('blob', ?)", (buffer("blob"),))
val = cur.fetchone()[0]
self.failUnlessEqual(val, 1)
def CheckAggrCheckAggrSum(self):
cur = self.con.cursor()
cur.execute("delete from test")
cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)])
cur.execute("select mysum(i) from test")
val = cur.fetchone()[0]
self.failUnlessEqual(val, 60)
def suite():
function_suite = unittest.makeSuite(FunctionTests, "Check")
aggregate_suite = unittest.makeSuite(AggregateTests, "Check")
return unittest.TestSuite((function_suite, aggregate_suite))
def test():
runner = unittest.TextTestRunner()
runner.run(suite())
if __name__ == "__main__":
test()

16
Lib/test/regrtest.py
View File

@ -741,6 +741,7 @@ _expectations = {
test_pwd
test_resource
test_signal
test_sqlite
test_sunaudiodev
test_threadsignals
test_timing
@ -763,6 +764,7 @@ _expectations = {
test_nis
test_ntpath
test_ossaudiodev
test_sqlite
test_sunaudiodev
""",
'mac':
@ -802,6 +804,7 @@ _expectations = {
test_pwd
test_resource
test_signal
test_sqlite
test_sunaudiodev
test_sundry
test_tarfile
@ -826,6 +829,7 @@ _expectations = {
test_openpty
test_pyexpat
test_sax
test_sqlite
test_sunaudiodev
test_sundry
""",
@ -848,6 +852,7 @@ _expectations = {
test_openpty
test_pyexpat
test_sax
test_sqlite
test_sunaudiodev
test_sundry
""",
@ -875,6 +880,7 @@ _expectations = {
test_pyexpat
test_queue
test_sax
test_sqlite
test_sunaudiodev
test_sundry
test_thread
@ -915,6 +921,7 @@ _expectations = {
test_pty
test_pwd
test_strop
test_sqlite
test_sunaudiodev
test_sundry
test_thread
@ -944,6 +951,7 @@ _expectations = {
test_ntpath
test_ossaudiodev
test_poll
test_sqlite
test_sunaudiodev
""",
'sunos5':
@ -962,6 +970,7 @@ _expectations = {
test_imgfile
test_linuxaudiodev
test_openpty
test_sqlite
test_zipfile
test_zlib
""",
@ -988,6 +997,7 @@ _expectations = {
test_openpty
test_pyexpat
test_sax
test_sqlite
test_sunaudiodev
test_zipfile
test_zlib
@ -1013,6 +1023,7 @@ _expectations = {
test_poll
test_popen2
test_resource
test_sqlite
test_sunaudiodev
""",
'cygwin':
@ -1034,6 +1045,7 @@ _expectations = {
test_nis
test_ossaudiodev
test_socketserver
test_sqlite
test_sunaudiodev
""",
'os2emx':
@ -1060,6 +1072,7 @@ _expectations = {
test_pty
test_resource
test_signal
test_sqlite
test_sunaudiodev
""",
'freebsd4':
@ -1086,6 +1099,7 @@ _expectations = {
test_scriptpackages
test_socket_ssl
test_socketserver
test_sqlite
test_sunaudiodev
test_tcl
test_timeout
@ -1115,6 +1129,7 @@ _expectations = {
test_macostools
test_nis
test_ossaudiodev
test_sqlite
test_sunaudiodev
test_tcl
test_winreg
@ -1147,6 +1162,7 @@ _expectations = {
test_plistlib
test_scriptpackages
test_tcl
test_sqlite
test_sunaudiodev
test_unicode_file
test_winreg

16
Lib/test/test_sqlite.py Normal file
View File

@ -0,0 +1,16 @@
from test.test_support import run_unittest, TestSkipped
import unittest
try:
import _sqlite3
except ImportError:
raise TestSkipped('no sqlite available')
from sqlite3.test import (dbapi, types, userfunctions,
factory, transactions)
def test_main():
run_unittest(dbapi.suite(), types.suite(), userfunctions.suite(),
factory.suite(), transactions.suite())
if __name__ == "__main__":
test_main()

5
Misc/NEWS
View File

@ -489,6 +489,11 @@ Extension Modules
Library
-------
- Added the sqlite3 package. This is based on pysqlite2.1.3, and provides
a DB-API interface in the standard library. You'll need sqlite 3.2.2 or
later to build this - if you have an earlier version, the C extension
module will not be built.
- Bug #1460340: ``random.sample(dict)`` failed in various ways. Dicts
aren't officially supported here, and trying to use them will probably
raise an exception some day. But dicts have been allowed, and "mostly

40
Modules/_sqlite/adapters.c Normal file
View File

@ -0,0 +1,40 @@
/* adapters.c - default adapters
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "util.h"
#include "module.h"
#include "adapters.h"
/* dummy, will be implemented in a later version */
PyObject* adapt_date(PyObject* self, PyObject* args, PyObject* kwargs)
{
Py_INCREF(Py_None);
return Py_None;
}
PyObject* adapt_datetime(PyObject* self, PyObject* args, PyObject* kwargs)
{
Py_INCREF(Py_None);
return Py_None;
}

33
Modules/_sqlite/adapters.h Normal file
View File

@ -0,0 +1,33 @@
/* adapters.h - default adapters
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_ADAPTERS_H
#define PYSQLITE_ADAPTERS_H
#include "Python.h"
#include "pythread.h"
#include "sqlite3.h"
PyObject* adapt_date(PyObject* self, PyObject* args, PyObject* kwargs);
PyObject* adapt_datetime(PyObject* self, PyObject* args, PyObject* kwargs);
#endif

343
Modules/_sqlite/cache.c Normal file
View File

@ -0,0 +1,343 @@
/* cache .c - a LRU cache
*
* Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "cache.h"
/* only used internally */
Node* new_node(PyObject* key, PyObject* data)
{
Node* node;
node = (Node*) (NodeType.tp_alloc(&NodeType, 0));
/*node = PyObject_New(Node, &NodeType);*/
if (!node) {
return NULL;
}
Py_INCREF(key);
node->key = key;
Py_INCREF(data);
node->data = data;
node->prev = NULL;
node->next = NULL;
return node;
}
void node_dealloc(Node* self)
{
Py_DECREF(self->key);
Py_DECREF(self->data);
self->ob_type->tp_free((PyObject*)self);
}
int cache_init(Cache* self, PyObject* args, PyObject* kwargs)
{
PyObject* factory;
int size = 10;
self->factory = NULL;
if (!PyArg_ParseTuple(args, "O|i", &factory, &size))
{
return -1;
}
if (size < 5) {
size = 5;
}
self->size = size;
self->first = NULL;
self->last = NULL;
self->mapping = PyDict_New();
Py_INCREF(factory);
self->factory = factory;
self->decref_factory = 1;
return 0;
}
void cache_dealloc(Cache* self)
{
Node* node;
Node* delete_node;
if (!self->factory) {
/* constructor failed, just get out of here */
return;
}
node = self->first;
while (node) {
delete_node = node;
node = node->next;
Py_DECREF(delete_node);
}
if (self->decref_factory) {
Py_DECREF(self->factory);
}
Py_DECREF(self->mapping);
self->ob_type->tp_free((PyObject*)self);
}
PyObject* cache_get(Cache* self, PyObject* args)
{
PyObject* key;
Node* node;
Node* ptr;
PyObject* data;
if (!PyArg_ParseTuple(args, "O", &key))
{
return NULL;
}
node = (Node*)PyDict_GetItem(self->mapping, key);
if (node) {
node->count++;
if (node->prev && node->count > node->prev->count) {
ptr = node->prev;
while (ptr->prev && node->count > ptr->prev->count) {
ptr = ptr->prev;
}
if (node->next) {
node->next->prev = node->prev;
} else {
self->last = node->prev;
}
if (node->prev) {
node->prev->next = node->next;
}
if (ptr->prev) {
ptr->prev->next = node;
} else {
self->first = node;
}
node->next = ptr;
node->prev = ptr->prev;
if (!node->prev) {
self->first = node;
}
ptr->prev = node;
}
} else {
if (PyDict_Size(self->mapping) == self->size) {
if (self->last) {
node = self->last;
PyDict_DelItem(self->mapping, self->last->key);
if (node->prev) {
node->prev->next = NULL;
}
self->last = node->prev;
node->prev = NULL;
Py_DECREF(node);
}
}
data = PyObject_CallFunction(self->factory, "O", key);
if (!data) {
return NULL;
}
node = new_node(key, data);
node->prev = self->last;
Py_DECREF(data);
if (self->last) {
self->last->next = node;
} else {
self->first = node;
}
self->last = node;
PyDict_SetItem(self->mapping, key, (PyObject*)node);
}
Py_INCREF(node->data);
return node->data;
}
PyObject* cache_display(Cache* self, PyObject* args)
{
Node* ptr;
PyObject* prevkey;
PyObject* nextkey;
PyObject* fmt_args;
PyObject* template;
PyObject* display_str;
ptr = self->first;
while (ptr) {
if (ptr->prev) {
prevkey = ptr->prev->key;
} else {
prevkey = Py_None;
}
Py_INCREF(prevkey);
if (ptr->next) {
nextkey = ptr->next->key;
} else {
nextkey = Py_None;
}
Py_INCREF(nextkey);
fmt_args = Py_BuildValue("OOO", prevkey, ptr->key, nextkey);
template = PyString_FromString("%s <- %s ->%s\n");
display_str = PyString_Format(template, fmt_args);
Py_DECREF(template);
Py_DECREF(fmt_args);
PyObject_Print(display_str, stdout, Py_PRINT_RAW);
Py_DECREF(display_str);
Py_DECREF(prevkey);
Py_DECREF(nextkey);
ptr = ptr->next;
}
Py_INCREF(Py_None);
return Py_None;
}
static PyMethodDef cache_methods[] = {
{"get", (PyCFunction)cache_get, METH_VARARGS,
PyDoc_STR("Gets an entry from the cache.")},
{"display", (PyCFunction)cache_display, METH_NOARGS,
PyDoc_STR("For debugging only.")},
{NULL, NULL}
};
PyTypeObject NodeType = {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"pysqlite2.dbapi2.Node", /* tp_name */
sizeof(Node), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)node_dealloc, /* tp_dealloc */
0, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
0, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
0, /* tp_hash */
0, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, /* tp_flags */
0, /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
0, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)0, /* tp_init */
0, /* tp_alloc */
0, /* tp_new */
0 /* tp_free */
};
PyTypeObject CacheType = {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"pysqlite2.dbapi2.Cache", /* tp_name */
sizeof(Cache), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)cache_dealloc, /* tp_dealloc */
0, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
0, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
0, /* tp_hash */
0, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, /* tp_flags */
0, /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
cache_methods, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)cache_init, /* tp_init */
0, /* tp_alloc */
0, /* tp_new */
0 /* tp_free */
};
extern int cache_setup_types(void)
{
int rc;
NodeType.tp_new = PyType_GenericNew;
CacheType.tp_new = PyType_GenericNew;
rc = PyType_Ready(&NodeType);
if (rc < 0) {
return rc;
}
rc = PyType_Ready(&CacheType);
return rc;
}

61
Modules/_sqlite/cache.h Normal file
View File

@ -0,0 +1,61 @@
/* cache.h - definitions for the LRU cache
*
* Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_CACHE_H
#define PYSQLITE_CACHE_H
#include "Python.h"
typedef struct _Node
{
PyObject_HEAD
PyObject* key;
PyObject* data;
long count;
struct _Node* prev;
struct _Node* next;
} Node;
typedef struct
{
PyObject_HEAD
int size;
PyObject* mapping;
PyObject* factory;
Node* first;
Node* last;
int decref_factory;
} Cache;
extern PyTypeObject NodeType;
extern PyTypeObject CacheType;
int node_init(Node* self, PyObject* args, PyObject* kwargs);
void node_dealloc(Node* self);
int cache_init(Cache* self, PyObject* args, PyObject* kwargs);
void cache_dealloc(Cache* self);
PyObject* cache_get(Cache* self, PyObject* args);
int cache_setup_types(void);
#endif

922
Modules/_sqlite/connection.c Normal file
View File

@ -0,0 +1,922 @@
/* connection.c - the connection type
*
* Copyright (C) 2004-2006 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "cache.h"
#include "module.h"
#include "connection.h"
#include "statement.h"
#include "cursor.h"
#include "prepare_protocol.h"
#include "util.h"
#include "pythread.h"
static int connection_set_isolation_level(Connection* self, PyObject* isolation_level);
int connection_init(Connection* self, PyObject* args, PyObject* kwargs)
{
static char *kwlist[] = {"database", "timeout", "detect_types", "isolation_level", "check_same_thread", "factory", "cached_statements", NULL, NULL};
char* database;
int detect_types = 0;
PyObject* isolation_level = NULL;
PyObject* factory = NULL;
int check_same_thread = 1;
int cached_statements = 100;
double timeout = 5.0;
int rc;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|diOiOi", kwlist,
&database, &timeout, &detect_types, &isolation_level, &check_same_thread, &factory, &cached_statements))
{
return -1;
}
self->begin_statement = NULL;
self->statement_cache = NULL;
Py_INCREF(Py_None);
self->row_factory = Py_None;
Py_INCREF(&PyUnicode_Type);
self->text_factory = (PyObject*)&PyUnicode_Type;
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_open(database, &self->db);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK) {
_seterror(self->db);
return -1;
}
if (!isolation_level) {
isolation_level = PyString_FromString("");
} else {
Py_INCREF(isolation_level);
}
self->isolation_level = NULL;
connection_set_isolation_level(self, isolation_level);
Py_DECREF(isolation_level);
self->statement_cache = (Cache*)PyObject_CallFunction((PyObject*)&CacheType, "Oi", self, cached_statements);
if (PyErr_Occurred()) {
return -1;
}
/* By default, the Cache class INCREFs the factory in its initializer, and
* decrefs it in its deallocator method. Since this would create a circular
* reference here, we're breaking it by decrementing self, and telling the
* cache class to not decref the factory (self) in its deallocator.
*/
self->statement_cache->decref_factory = 0;
Py_DECREF(self);
self->inTransaction = 0;
self->detect_types = detect_types;
self->timeout = timeout;
(void)sqlite3_busy_timeout(self->db, (int)(timeout*1000));
self->thread_ident = PyThread_get_thread_ident();
self->check_same_thread = check_same_thread;
self->function_pinboard = PyDict_New();
self->Warning = Warning;
self->Error = Error;
self->InterfaceError = InterfaceError;
self->DatabaseError = DatabaseError;
self->DataError = DataError;
self->OperationalError = OperationalError;
self->IntegrityError = IntegrityError;
self->InternalError = InternalError;
self->ProgrammingError = ProgrammingError;
self->NotSupportedError = NotSupportedError;
return 0;
}
void flush_statement_cache(Connection* self)
{
Node* node;
Statement* statement;
node = self->statement_cache->first;
while (node) {
statement = (Statement*)(node->data);
(void)statement_finalize(statement);
node = node->next;
}
Py_DECREF(self->statement_cache);
self->statement_cache = (Cache*)PyObject_CallFunction((PyObject*)&CacheType, "O", self);
Py_DECREF(self);
self->statement_cache->decref_factory = 0;
}
void reset_all_statements(Connection* self)
{
Node* node;
Statement* statement;
node = self->statement_cache->first;
while (node) {
statement = (Statement*)(node->data);
(void)statement_reset(statement);
node = node->next;
}
}
void connection_dealloc(Connection* self)
{
Py_XDECREF(self->statement_cache);
/* Clean up if user has not called .close() explicitly. */
if (self->db) {
Py_BEGIN_ALLOW_THREADS
sqlite3_close(self->db);
Py_END_ALLOW_THREADS
}
if (self->begin_statement) {
PyMem_Free(self->begin_statement);
}
Py_XDECREF(self->isolation_level);
Py_XDECREF(self->function_pinboard);
Py_XDECREF(self->row_factory);
Py_XDECREF(self->text_factory);
self->ob_type->tp_free((PyObject*)self);
}
PyObject* connection_cursor(Connection* self, PyObject* args, PyObject* kwargs)
{
static char *kwlist[] = {"factory", NULL, NULL};
PyObject* factory = NULL;
PyObject* cursor;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|O", kwlist,
&factory)) {
return NULL;
}
if (!check_thread(self) || !check_connection(self)) {
return NULL;
}
if (factory == NULL) {
factory = (PyObject*)&CursorType;
}
cursor = PyObject_CallFunction(factory, "O", self);
if (cursor && self->row_factory != Py_None) {
Py_XDECREF(((Cursor*)cursor)->row_factory);
Py_INCREF(self->row_factory);
((Cursor*)cursor)->row_factory = self->row_factory;
}
return cursor;
}
PyObject* connection_close(Connection* self, PyObject* args)
{
int rc;
if (!check_thread(self)) {
return NULL;
}
flush_statement_cache(self);
if (self->db) {
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_close(self->db);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK) {
_seterror(self->db);
return NULL;
} else {
self->db = NULL;
}
}
Py_INCREF(Py_None);
return Py_None;
}
/*
* Checks if a connection object is usable (i. e. not closed).
*
* 0 => error; 1 => ok
*/
int check_connection(Connection* con)
{
if (!con->db) {
PyErr_SetString(ProgrammingError, "Cannot operate on a closed database.");
return 0;
} else {
return 1;
}
}
PyObject* _connection_begin(Connection* self)
{
int rc;
const char* tail;
sqlite3_stmt* statement;
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_prepare(self->db, self->begin_statement, -1, &statement, &tail);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK) {
_seterror(self->db);
goto error;
}
rc = _sqlite_step_with_busyhandler(statement, self);
if (rc == SQLITE_DONE) {
self->inTransaction = 1;
} else {
_seterror(self->db);
}
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_finalize(statement);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK && !PyErr_Occurred()) {
_seterror(self->db);
}
error:
if (PyErr_Occurred()) {
return NULL;
} else {
Py_INCREF(Py_None);
return Py_None;
}
}
PyObject* connection_commit(Connection* self, PyObject* args)
{
int rc;
const char* tail;
sqlite3_stmt* statement;
if (!check_thread(self) || !check_connection(self)) {
return NULL;
}
if (self->inTransaction) {
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_prepare(self->db, "COMMIT", -1, &statement, &tail);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK) {
_seterror(self->db);
goto error;
}
rc = _sqlite_step_with_busyhandler(statement, self);
if (rc == SQLITE_DONE) {
self->inTransaction = 0;
} else {
_seterror(self->db);
}
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_finalize(statement);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK && !PyErr_Occurred()) {
_seterror(self->db);
}
}
error:
if (PyErr_Occurred()) {
return NULL;
} else {
Py_INCREF(Py_None);
return Py_None;
}
}
PyObject* connection_rollback(Connection* self, PyObject* args)
{
int rc;
const char* tail;
sqlite3_stmt* statement;
if (!check_thread(self) || !check_connection(self)) {
return NULL;
}
if (self->inTransaction) {
reset_all_statements(self);
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_prepare(self->db, "ROLLBACK", -1, &statement, &tail);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK) {
_seterror(self->db);
goto error;
}
rc = _sqlite_step_with_busyhandler(statement, self);
if (rc == SQLITE_DONE) {
self->inTransaction = 0;
} else {
_seterror(self->db);
}
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_finalize(statement);
Py_END_ALLOW_THREADS
if (rc != SQLITE_OK && !PyErr_Occurred()) {
_seterror(self->db);
}
}
error:
if (PyErr_Occurred()) {
return NULL;
} else {
Py_INCREF(Py_None);
return Py_None;
}
}
void _set_result(sqlite3_context* context, PyObject* py_val)
{
long longval;
const char* buffer;
int buflen;
PyObject* stringval;
if (PyErr_Occurred()) {
/* Errors in callbacks are ignored, and we return NULL */
PyErr_Clear();
sqlite3_result_null(context);
} else if (py_val == Py_None) {
sqlite3_result_null(context);
} else if (PyInt_Check(py_val)) {
longval = PyInt_AsLong(py_val);
/* TODO: investigate what to do with range overflows - long vs. long long */
sqlite3_result_int64(context, (PY_LONG_LONG)longval);
} else if (PyFloat_Check(py_val)) {
sqlite3_result_double(context, PyFloat_AsDouble(py_val));
} else if (PyBuffer_Check(py_val)) {
if (PyObject_AsCharBuffer(py_val, &buffer, &buflen) != 0) {
PyErr_SetString(PyExc_ValueError, "could not convert BLOB to buffer");
}
sqlite3_result_blob(context, buffer, buflen, SQLITE_TRANSIENT);
} else if (PyString_Check(py_val)) {
sqlite3_result_text(context, PyString_AsString(py_val), -1, SQLITE_TRANSIENT);
} else if (PyUnicode_Check(py_val)) {
stringval = PyUnicode_AsUTF8String(py_val);
sqlite3_result_text(context, PyString_AsString(stringval), -1, SQLITE_TRANSIENT);
Py_DECREF(stringval);
} else {
/* TODO: raise error */
}
}
PyObject* _build_py_params(sqlite3_context *context, int argc, sqlite3_value** argv)
{
PyObject* args;
int i;
sqlite3_value* cur_value;
PyObject* cur_py_value;
const char* val_str;
PY_LONG_LONG val_int;
int buflen;
void* raw_buffer;
args = PyTuple_New(argc);
for (i = 0; i < argc; i++) {
cur_value = argv[i];
switch (sqlite3_value_type(argv[i])) {
case SQLITE_INTEGER:
val_int = sqlite3_value_int64(cur_value);
cur_py_value = PyInt_FromLong((long)val_int);
break;
case SQLITE_FLOAT:
cur_py_value = PyFloat_FromDouble(sqlite3_value_double(cur_value));
break;
case SQLITE_TEXT:
val_str = (const char*)sqlite3_value_text(cur_value);
cur_py_value = PyUnicode_DecodeUTF8(val_str, strlen(val_str), NULL);
/* TODO: have a way to show errors here */
if (!cur_py_value) {
Py_INCREF(Py_None);
cur_py_value = Py_None;
}
break;
case SQLITE_BLOB:
buflen = sqlite3_value_bytes(cur_value);
cur_py_value = PyBuffer_New(buflen);
if (!cur_py_value) {
/* TODO: error */
}
if (PyObject_AsWriteBuffer(cur_py_value, &raw_buffer, &buflen)) {
/* TODO: error */
}
memcpy(raw_buffer, sqlite3_value_blob(cur_value), buflen);
break;
case SQLITE_NULL:
default:
Py_INCREF(Py_None);
cur_py_value = Py_None;
}
PyTuple_SetItem(args, i, cur_py_value);
}
return args;
}
void _func_callback(sqlite3_context* context, int argc, sqlite3_value** argv)
{
PyObject* args;
PyObject* py_func;
PyObject* py_retval;
PyGILState_STATE threadstate;
threadstate = PyGILState_Ensure();
py_func = (PyObject*)sqlite3_user_data(context);
args = _build_py_params(context, argc, argv);
py_retval = PyObject_CallObject(py_func, args);
Py_DECREF(args);
_set_result(context, py_retval);
Py_XDECREF(py_retval);
PyGILState_Release(threadstate);
}
static void _step_callback(sqlite3_context *context, int argc, sqlite3_value** params)
{
PyObject* args;
PyObject* function_result;
PyObject* aggregate_class;
PyObject** aggregate_instance;
PyObject* stepmethod;
PyGILState_STATE threadstate;
threadstate = PyGILState_Ensure();
aggregate_class = (PyObject*)sqlite3_user_data(context);
aggregate_instance = (PyObject**)sqlite3_aggregate_context(context, sizeof(PyObject*));
if (*aggregate_instance == 0) {
*aggregate_instance = PyObject_CallFunction(aggregate_class, "");
if (PyErr_Occurred())
{
PyErr_Clear();
*aggregate_instance = 0;
PyGILState_Release(threadstate);
return;
}
}
stepmethod = PyObject_GetAttrString(*aggregate_instance, "step");
if (!stepmethod)
{
PyGILState_Release(threadstate);
return;
}
args = _build_py_params(context, argc, params);
function_result = PyObject_CallObject(stepmethod, args);
Py_DECREF(args);
Py_DECREF(stepmethod);
if (function_result == NULL) {
PyErr_Clear();
} else {
Py_DECREF(function_result);
}
PyGILState_Release(threadstate);
}
void _final_callback(sqlite3_context* context)
{
PyObject* args;
PyObject* function_result;
PyObject** aggregate_instance;
PyObject* aggregate_class;
PyObject* finalizemethod;
PyGILState_STATE threadstate;
threadstate = PyGILState_Ensure();
aggregate_class = (PyObject*)sqlite3_user_data(context);
aggregate_instance = (PyObject**)sqlite3_aggregate_context(context, sizeof(PyObject*));
if (!*aggregate_instance) {
/* this branch is executed if there was an exception in the aggregate's
* __init__ */
PyGILState_Release(threadstate);
return;
}
finalizemethod = PyObject_GetAttrString(*aggregate_instance, "finalize");
if (!finalizemethod) {
/*
PyErr_SetString(ProgrammingError, "finalize method missing");
goto error;
*/
Py_INCREF(Py_None);
function_result = Py_None;
} else {
args = PyTuple_New(0);
function_result = PyObject_CallObject(finalizemethod, args);
Py_DECREF(args);
Py_DECREF(finalizemethod);
}
_set_result(context, function_result);
Py_XDECREF(*aggregate_instance);
Py_XDECREF(function_result);
PyGILState_Release(threadstate);
}
PyObject* connection_create_function(Connection* self, PyObject* args, PyObject* kwargs)
{
static char *kwlist[] = {"name", "narg", "func", NULL, NULL};
PyObject* func;
char* name;
int narg;
int rc;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "siO", kwlist,
&name, &narg, &func))
{
return NULL;
}
rc = sqlite3_create_function(self->db, name, narg, SQLITE_UTF8, (void*)func, _func_callback, NULL, NULL);
PyDict_SetItem(self->function_pinboard, func, Py_None);
Py_INCREF(Py_None);
return Py_None;
}
PyObject* connection_create_aggregate(Connection* self, PyObject* args, PyObject* kwargs)
{
PyObject* aggregate_class;
int n_arg;
char* name;
static char *kwlist[] = { "name", "n_arg", "aggregate_class", NULL };
int rc;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "siO:create_aggregate",
kwlist, &name, &n_arg, &aggregate_class)) {
return NULL;
}
rc = sqlite3_create_function(self->db, name, n_arg, SQLITE_UTF8, (void*)aggregate_class, 0, &_step_callback, &_final_callback);
if (rc != SQLITE_OK) {
_seterror(self->db);
return NULL;
} else {
PyDict_SetItem(self->function_pinboard, aggregate_class, Py_None);
Py_INCREF(Py_None);
return Py_None;
}
}
int check_thread(Connection* self)
{
if (self->check_same_thread) {
if (PyThread_get_thread_ident() != self->thread_ident) {
PyErr_Format(ProgrammingError,
"SQLite objects created in a thread can only be used in that same thread."
"The object was created in thread id %ld and this is thread id %ld",
self->thread_ident, PyThread_get_thread_ident());
return 0;
}
}
return 1;
}
static PyObject* connection_get_isolation_level(Connection* self, void* unused)
{
Py_INCREF(self->isolation_level);
return self->isolation_level;
}
static int connection_set_isolation_level(Connection* self, PyObject* isolation_level)
{
PyObject* empty;
PyObject* res;
PyObject* begin_statement;
Py_XDECREF(self->isolation_level);
if (isolation_level == Py_None) {
Py_INCREF(Py_None);
self->begin_statement = NULL;
self->isolation_level = Py_None;
empty = PyTuple_New(0);
res = connection_commit(self, empty);
Py_DECREF(empty);
Py_DECREF(res);
self->inTransaction = 0;
} else {
Py_INCREF(isolation_level);
self->isolation_level = isolation_level;
begin_statement = PyString_FromString("BEGIN ");
if (!begin_statement) {
return -1;
}
PyString_Concat(&begin_statement, isolation_level);
if (!begin_statement) {
return -1;
}
self->begin_statement = PyMem_Malloc(PyString_Size(begin_statement) + 2);
if (!self->begin_statement) {
return -1;
}
strcpy(self->begin_statement, PyString_AsString(begin_statement));
Py_DECREF(begin_statement);
}
return 0;
}
PyObject* connection_call(Connection* self, PyObject* args, PyObject* kwargs)
{
PyObject* sql;
Statement* statement;
int rc;
if (!PyArg_ParseTuple(args, "O", &sql)) {
return NULL;
}
statement = PyObject_New(Statement, &StatementType);
if (!statement) {
return NULL;
}
rc = statement_create(statement, self, sql);
if (rc != SQLITE_OK) {
if (rc == PYSQLITE_TOO_MUCH_SQL) {
PyErr_SetString(Warning, "You can only execute one statement at a time.");
} else if (rc == PYSQLITE_SQL_WRONG_TYPE) {
PyErr_SetString(Warning, "SQL is of wrong type. Must be string or unicode.");
} else {
_seterror(self->db);
}
Py_DECREF(statement);
statement = 0;
}
return (PyObject*)statement;
}
PyObject* connection_execute(Connection* self, PyObject* args, PyObject* kwargs)
{
PyObject* cursor = 0;
PyObject* result = 0;
PyObject* method = 0;
cursor = PyObject_CallMethod((PyObject*)self, "cursor", "");
if (!cursor) {
goto error;
}
method = PyObject_GetAttrString(cursor, "execute");
if (!method) {
Py_DECREF(cursor);
cursor = 0;
goto error;
}
result = PyObject_CallObject(method, args);
if (!result) {
Py_DECREF(cursor);
cursor = 0;
}
error:
Py_XDECREF(result);
Py_XDECREF(method);
return cursor;
}
PyObject* connection_executemany(Connection* self, PyObject* args, PyObject* kwargs)
{
PyObject* cursor = 0;
PyObject* result = 0;
PyObject* method = 0;
cursor = PyObject_CallMethod((PyObject*)self, "cursor", "");
if (!cursor) {
goto error;
}
method = PyObject_GetAttrString(cursor, "executemany");
if (!method) {
Py_DECREF(cursor);
cursor = 0;
goto error;
}
result = PyObject_CallObject(method, args);
if (!result) {
Py_DECREF(cursor);
cursor = 0;
}
error:
Py_XDECREF(result);
Py_XDECREF(method);
return cursor;
}
PyObject* connection_executescript(Connection* self, PyObject* args, PyObject* kwargs)
{
PyObject* cursor = 0;
PyObject* result = 0;
PyObject* method = 0;
cursor = PyObject_CallMethod((PyObject*)self, "cursor", "");
if (!cursor) {
goto error;
}
method = PyObject_GetAttrString(cursor, "executescript");
if (!method) {
Py_DECREF(cursor);
cursor = 0;
goto error;
}
result = PyObject_CallObject(method, args);
if (!result) {
Py_DECREF(cursor);
cursor = 0;
}
error:
Py_XDECREF(result);
Py_XDECREF(method);
return cursor;
}
static char connection_doc[] =
PyDoc_STR("<missing docstring>");
static PyGetSetDef connection_getset[] = {
{"isolation_level", (getter)connection_get_isolation_level, (setter)connection_set_isolation_level},
{NULL}
};
static PyMethodDef connection_methods[] = {
{"cursor", (PyCFunction)connection_cursor, METH_VARARGS|METH_KEYWORDS,
PyDoc_STR("Return a cursor for the connection.")},
{"close", (PyCFunction)connection_close, METH_NOARGS,
PyDoc_STR("Closes the connection.")},
{"commit", (PyCFunction)connection_commit, METH_NOARGS,
PyDoc_STR("Commit the current transaction.")},
{"rollback", (PyCFunction)connection_rollback, METH_NOARGS,
PyDoc_STR("Roll back the current transaction.")},
{"create_function", (PyCFunction)connection_create_function, METH_VARARGS|METH_KEYWORDS,
PyDoc_STR("Creates a new function. Non-standard.")},
{"create_aggregate", (PyCFunction)connection_create_aggregate, METH_VARARGS|METH_KEYWORDS,
PyDoc_STR("Creates a new aggregate. Non-standard.")},
{"execute", (PyCFunction)connection_execute, METH_VARARGS,
PyDoc_STR("Executes a SQL statement. Non-standard.")},
{"executemany", (PyCFunction)connection_executemany, METH_VARARGS,
PyDoc_STR("Repeatedly executes a SQL statement. Non-standard.")},
{"executescript", (PyCFunction)connection_executescript, METH_VARARGS,
PyDoc_STR("Executes a multiple SQL statements at once. Non-standard.")},
{NULL, NULL}
};
static struct PyMemberDef connection_members[] =
{
{"Warning", T_OBJECT, offsetof(Connection, Warning), RO},
{"Error", T_OBJECT, offsetof(Connection, Error), RO},
{"InterfaceError", T_OBJECT, offsetof(Connection, InterfaceError), RO},
{"DatabaseError", T_OBJECT, offsetof(Connection, DatabaseError), RO},
{"DataError", T_OBJECT, offsetof(Connection, DataError), RO},
{"OperationalError", T_OBJECT, offsetof(Connection, OperationalError), RO},
{"IntegrityError", T_OBJECT, offsetof(Connection, IntegrityError), RO},
{"InternalError", T_OBJECT, offsetof(Connection, InternalError), RO},
{"ProgrammingError", T_OBJECT, offsetof(Connection, ProgrammingError), RO},
{"NotSupportedError", T_OBJECT, offsetof(Connection, NotSupportedError), RO},
{"row_factory", T_OBJECT, offsetof(Connection, row_factory)},
{"text_factory", T_OBJECT, offsetof(Connection, text_factory)},
{NULL}
};
PyTypeObject ConnectionType = {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"pysqlite2.dbapi2.Connection", /* tp_name */
sizeof(Connection), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)connection_dealloc, /* tp_dealloc */
0, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
0, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
0, /* tp_hash */
(ternaryfunc)connection_call, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, /* tp_flags */
connection_doc, /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
connection_methods, /* tp_methods */
connection_members, /* tp_members */
connection_getset, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)connection_init, /* tp_init */
0, /* tp_alloc */
0, /* tp_new */
0 /* tp_free */
};
extern int connection_setup_types(void)
{
ConnectionType.tp_new = PyType_GenericNew;
return PyType_Ready(&ConnectionType);
}

103
Modules/_sqlite/connection.h Normal file
View File

@ -0,0 +1,103 @@
/* connection.h - definitions for the connection type
*
* Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_CONNECTION_H
#define PYSQLITE_CONNECTION_H
#include "Python.h"
#include "pythread.h"
#include "structmember.h"
#include "cache.h"
#include "module.h"
#include "sqlite3.h"
typedef struct
{
PyObject_HEAD
sqlite3* db;
int inTransaction;
int detect_types;
/* the timeout value in seconds for database locks */
double timeout;
/* for internal use in the timeout handler: when did the timeout handler
* first get called with count=0? */
double timeout_started;
/* None for autocommit, otherwise a PyString with the isolation level */
PyObject* isolation_level;
/* NULL for autocommit, otherwise a string with the BEGIN statment; will be
* freed in connection destructor */
char* begin_statement;
int check_same_thread;
long thread_ident;
Cache* statement_cache;
PyObject* row_factory;
PyObject* text_factory;
/* remember references to functions/classes used in
* create_function/create/aggregate, use these as dictionary keys, so we
* can keep the total system refcount constant by clearing that dictionary
* in connection_dealloc */
PyObject* function_pinboard;
/* Exception objects */
PyObject* Warning;
PyObject* Error;
PyObject* InterfaceError;
PyObject* DatabaseError;
PyObject* DataError;
PyObject* OperationalError;
PyObject* IntegrityError;
PyObject* InternalError;
PyObject* ProgrammingError;
PyObject* NotSupportedError;
} Connection;
extern PyTypeObject ConnectionType;
PyObject* connection_alloc(PyTypeObject* type, int aware);
void connection_dealloc(Connection* self);
PyObject* connection_cursor(Connection* self, PyObject* args, PyObject* kwargs);
PyObject* connection_close(Connection* self, PyObject* args);
PyObject* _connection_begin(Connection* self);
PyObject* connection_begin(Connection* self, PyObject* args);
PyObject* connection_commit(Connection* self, PyObject* args);
PyObject* connection_rollback(Connection* self, PyObject* args);
PyObject* connection_new(PyTypeObject* type, PyObject* args, PyObject* kw);
int connection_init(Connection* self, PyObject* args, PyObject* kwargs);
int check_thread(Connection* self);
int check_connection(Connection* con);
int connection_setup_types(void);
#endif

40
Modules/_sqlite/converters.c Normal file
View File

@ -0,0 +1,40 @@
/* converters.c - default converters
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "util.h"
#include "module.h"
#include "adapters.h"
/* dummy, will be implemented in a later version */
PyObject* convert_date(PyObject* self, PyObject* args, PyObject* kwargs)
{
Py_INCREF(Py_None);
return Py_None;
}
PyObject* convert_timestamp(PyObject* self, PyObject* args, PyObject* kwargs)
{
Py_INCREF(Py_None);
return Py_None;
}

33
Modules/_sqlite/converters.h Normal file
View File

@ -0,0 +1,33 @@
/* converters.h - default converters
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_CONVERTERS_H
#define PYSQLITE_CONVERTERS_H
#include "Python.h"
#include "pythread.h"
#include "sqlite3.h"
PyObject* convert_date(PyObject* self, PyObject* args, PyObject* kwargs);
PyObject* convert_timestamp(PyObject* self, PyObject* args, PyObject* kwargs);
#endif

1067
Modules/_sqlite/cursor.c Normal file
View File

File diff suppressed because it is too large Load Diff

71
Modules/_sqlite/cursor.h Normal file
View File

@ -0,0 +1,71 @@
/* cursor.h - definitions for the cursor type
*
* Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_CURSOR_H
#define PYSQLITE_CURSOR_H
#include "Python.h"
#include "statement.h"
#include "connection.h"
#include "module.h"
typedef struct
{
PyObject_HEAD
Connection* connection;
PyObject* description;
PyObject* row_cast_map;
int arraysize;
PyObject* lastrowid;
PyObject* rowcount;
PyObject* row_factory;
Statement* statement;
/* the next row to be returned, NULL if no next row available */
PyObject* next_row;
} Cursor;
typedef enum {
STATEMENT_INVALID, STATEMENT_INSERT, STATEMENT_DELETE,
STATEMENT_UPDATE, STATEMENT_REPLACE, STATEMENT_SELECT,
STATEMENT_OTHER
} StatementKind;
extern PyTypeObject CursorType;
int cursor_init(Cursor* self, PyObject* args, PyObject* kwargs);
void cursor_dealloc(Cursor* self);
PyObject* cursor_execute(Cursor* self, PyObject* args);
PyObject* cursor_executemany(Cursor* self, PyObject* args);
PyObject* cursor_getiter(Cursor *self);
PyObject* cursor_iternext(Cursor *self);
PyObject* cursor_fetchone(Cursor* self, PyObject* args);
PyObject* cursor_fetchmany(Cursor* self, PyObject* args);
PyObject* cursor_fetchall(Cursor* self, PyObject* args);
PyObject* pysqlite_noop(Connection* self, PyObject* args);
PyObject* cursor_close(Cursor* self, PyObject* args);
int cursor_setup_types(void);
#define UNKNOWN (-1)
#endif

141
Modules/_sqlite/microprotocols.c Normal file
View File

@ -0,0 +1,141 @@
/* microprotocols.c - minimalist and non-validating protocols implementation
*
* Copyright (C) 2003-2004 Federico Di Gregorio <fog@debian.org>
*
* This file is part of psycopg and was adapted for pysqlite. Federico Di
* Gregorio gave the permission to use it within pysqlite under the following
* license:
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include <Python.h>
#include <structmember.h>
#include "cursor.h"
#include "microprotocols.h"
#include "prepare_protocol.h"
/** the adapters registry **/
PyObject *psyco_adapters;
/* microprotocols_init - initialize the adapters dictionary */
int
microprotocols_init(PyObject *dict)
{
/* create adapters dictionary and put it in module namespace */
if ((psyco_adapters = PyDict_New()) == NULL) {
return -1;
}
PyDict_SetItemString(dict, "adapters", psyco_adapters);
return 0;
}
/* microprotocols_add - add a reverse type-caster to the dictionary */
int
microprotocols_add(PyTypeObject *type, PyObject *proto, PyObject *cast)
{
PyObject* key;
if (proto == NULL) proto = (PyObject*)&SQLitePrepareProtocolType;
/*
Dprintf("microprotocols_add: cast %p for (%s, ?)",
cast, type->tp_name);
*/
key = Py_BuildValue("(OO)", (PyObject*)type, proto);
PyDict_SetItem(psyco_adapters, key, cast);
Py_DECREF(key);
return 0;
}
/* microprotocols_adapt - adapt an object to the built-in protocol */
PyObject *
microprotocols_adapt(PyObject *obj, PyObject *proto, PyObject *alt)
{
PyObject *adapter, *key;
/* we don't check for exact type conformance as specified in PEP 246
because the SQLitePrepareProtocolType type is abstract and there is no
way to get a quotable object to be its instance */
/* look for an adapter in the registry */
key = Py_BuildValue("(OO)", (PyObject*)obj->ob_type, proto);
adapter = PyDict_GetItem(psyco_adapters, key);
Py_DECREF(key);
if (adapter) {
PyObject *adapted = PyObject_CallFunctionObjArgs(adapter, obj, NULL);
return adapted;
}
/* try to have the protocol adapt this object*/
if (PyObject_HasAttrString(proto, "__adapt__")) {
PyObject *adapted = PyObject_CallMethod(proto, "__adapt__", "O", obj);
if (adapted) {
if (adapted != Py_None) {
return adapted;
} else {
Py_DECREF(adapted);
}
}
if (PyErr_Occurred() && !PyErr_ExceptionMatches(PyExc_TypeError))
return NULL;
}
/* and finally try to have the object adapt itself */
if (PyObject_HasAttrString(obj, "__conform__")) {
PyObject *adapted = PyObject_CallMethod(obj, "__conform__","O", proto);
if (adapted) {
if (adapted != Py_None) {
return adapted;
} else {
Py_DECREF(adapted);
}
}
if (PyErr_Occurred() && !PyErr_ExceptionMatches(PyExc_TypeError)) {
return NULL;
}
}
/* else set the right exception and return NULL */
PyErr_SetString(ProgrammingError, "can't adapt");
return NULL;
}
/** module-level functions **/
PyObject *
psyco_microprotocols_adapt(Cursor *self, PyObject *args)
{
PyObject *obj, *alt = NULL;
PyObject *proto = (PyObject*)&SQLitePrepareProtocolType;
if (!PyArg_ParseTuple(args, "O|OO", &obj, &proto, &alt)) return NULL;
return microprotocols_adapt(obj, proto, alt);
}

59
Modules/_sqlite/microprotocols.h Normal file
View File

@ -0,0 +1,59 @@
/* microprotocols.c - definitions for minimalist and non-validating protocols
*
* Copyright (C) 2003-2004 Federico Di Gregorio <fog@debian.org>
*
* This file is part of psycopg and was adapted for pysqlite. Federico Di
* Gregorio gave the permission to use it within pysqlite under the following
* license:
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PSYCOPG_MICROPROTOCOLS_H
#define PSYCOPG_MICROPROTOCOLS_H 1
#include <Python.h>
#ifdef __cplusplus
extern "C" {
#endif
/** adapters registry **/
extern PyObject *psyco_adapters;
/** the names of the three mandatory methods **/
#define MICROPROTOCOLS_GETQUOTED_NAME "getquoted"
#define MICROPROTOCOLS_GETSTRING_NAME "getstring"
#define MICROPROTOCOLS_GETBINARY_NAME "getbinary"
/** exported functions **/
/* used by module.c to init the microprotocols system */
extern int microprotocols_init(PyObject *dict);
extern int microprotocols_add(
PyTypeObject *type, PyObject *proto, PyObject *cast);
extern PyObject *microprotocols_adapt(
PyObject *obj, PyObject *proto, PyObject *alt);
extern PyObject *
psyco_microprotocols_adapt(Cursor* self, PyObject *args);
#define psyco_microprotocols_adapt_doc \
"adapt(obj, protocol, alternate) -> adapt obj to given protocol"
#endif /* !defined(PSYCOPG_MICROPROTOCOLS_H) */

290
Modules/_sqlite/module.c Normal file
View File

@ -0,0 +1,290 @@
/* module.c - the module itself
*
* Copyright (C) 2004-2006 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "connection.h"
#include "statement.h"
#include "cursor.h"
#include "cache.h"
#include "prepare_protocol.h"
#include "microprotocols.h"
#include "row.h"
#if SQLITE_VERSION_NUMBER >= 3003003
#define HAVE_SHARED_CACHE
#endif
/* static objects at module-level */
PyObject* Error, *Warning, *InterfaceError, *DatabaseError, *InternalError,
*OperationalError, *ProgrammingError, *IntegrityError, *DataError,
*NotSupportedError, *OptimizedUnicode;
PyObject* time_time;
PyObject* time_sleep;
PyObject* converters;
static PyObject* module_connect(PyObject* self, PyObject* args, PyObject*
kwargs)
{
/* Python seems to have no way of extracting a single keyword-arg at
* C-level, so this code is redundant with the one in connection_init in
* connection.c and must always be copied from there ... */
static char *kwlist[] = {"database", "timeout", "detect_types", "isolation_level", "check_same_thread", "factory", "cached_statements", NULL, NULL};
char* database;
int detect_types = 0;
PyObject* isolation_level;
PyObject* factory = NULL;
int check_same_thread = 1;
int cached_statements;
double timeout = 5.0;
PyObject* result;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|diOiOi", kwlist,
&database, &timeout, &detect_types, &isolation_level, &check_same_thread, &factory, &cached_statements))
{
return NULL;
}
if (factory == NULL) {
factory = (PyObject*)&ConnectionType;
}
result = PyObject_Call(factory, args, kwargs);
return result;
}
static PyObject* module_complete(PyObject* self, PyObject* args, PyObject*
kwargs)
{
static char *kwlist[] = {"statement", NULL, NULL};
char* statement;
PyObject* result;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s", kwlist, &statement))
{
return NULL;
}
if (sqlite3_complete(statement)) {
result = Py_True;
} else {
result = Py_False;
}
Py_INCREF(result);
return result;
}
#ifdef HAVE_SHARED_CACHE
static PyObject* module_enable_shared_cache(PyObject* self, PyObject* args, PyObject*
kwargs)
{
static char *kwlist[] = {"do_enable", NULL, NULL};
int do_enable;
int rc;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "i", kwlist, &do_enable))
{
return NULL;
}
rc = sqlite3_enable_shared_cache(do_enable);
if (rc != SQLITE_OK) {
PyErr_SetString(OperationalError, "Changing the shared_cache flag failed");
return NULL;
} else {
Py_INCREF(Py_None);
return Py_None;
}
}
#endif /* HAVE_SHARED_CACHE */
static PyObject* module_register_adapter(PyObject* self, PyObject* args, PyObject* kwargs)
{
PyTypeObject* type;
PyObject* caster;
if (!PyArg_ParseTuple(args, "OO", &type, &caster)) {
return NULL;
}
microprotocols_add(type, (PyObject*)&SQLitePrepareProtocolType, caster);
Py_INCREF(Py_None);
return Py_None;
}
static PyObject* module_register_converter(PyObject* self, PyObject* args, PyObject* kwargs)
{
PyObject* name;
PyObject* callable;
if (!PyArg_ParseTuple(args, "OO", &name, &callable)) {
return NULL;
}
PyDict_SetItem(converters, name, callable);
Py_INCREF(Py_None);
return Py_None;
}
void converters_init(PyObject* dict)
{
converters = PyDict_New();
PyDict_SetItemString(dict, "converters", converters);
}
static PyMethodDef module_methods[] = {
{"connect", (PyCFunction)module_connect, METH_VARARGS|METH_KEYWORDS, PyDoc_STR("Creates a connection.")},
{"complete_statement", (PyCFunction)module_complete, METH_VARARGS|METH_KEYWORDS, PyDoc_STR("Checks if a string contains a complete SQL statement.")},
#ifdef HAVE_SHARED_CACHE
{"enable_shared_cache", (PyCFunction)module_enable_shared_cache, METH_VARARGS|METH_KEYWORDS, PyDoc_STR("Enable or disable shared cache mode for the calling thread.")},
#endif
{"register_adapter", (PyCFunction)module_register_adapter, METH_VARARGS, PyDoc_STR("Registers an adapter with pysqlite's adapter registry.")},
{"register_converter", (PyCFunction)module_register_converter, METH_VARARGS, PyDoc_STR("Registers a converter with pysqlite.")},
{"adapt", (PyCFunction)psyco_microprotocols_adapt, METH_VARARGS, psyco_microprotocols_adapt_doc},
{NULL, NULL}
};
PyMODINIT_FUNC init_sqlite3(void)
{
PyObject *module, *dict;
PyObject* time_module;
module = Py_InitModule("_sqlite3", module_methods);
if (
(row_setup_types() < 0) ||
(cursor_setup_types() < 0) ||
(connection_setup_types() < 0) ||
(cache_setup_types() < 0) ||
(statement_setup_types() < 0) ||
(prepare_protocol_setup_types() < 0)
) {
return;
}
Py_INCREF(&ConnectionType);
PyModule_AddObject(module, "Connection", (PyObject*) &ConnectionType);
Py_INCREF(&CursorType);
PyModule_AddObject(module, "Cursor", (PyObject*) &CursorType);
Py_INCREF(&CacheType);
PyModule_AddObject(module, "Statement", (PyObject*)&StatementType);
Py_INCREF(&StatementType);
PyModule_AddObject(module, "Cache", (PyObject*) &CacheType);
Py_INCREF(&SQLitePrepareProtocolType);
PyModule_AddObject(module, "PrepareProtocol", (PyObject*) &SQLitePrepareProtocolType);
Py_INCREF(&RowType);
PyModule_AddObject(module, "Row", (PyObject*) &RowType);
if (!(dict = PyModule_GetDict(module)))
{
goto error;
}
/*** Create DB-API Exception hierarchy */
Error = PyErr_NewException("sqlite3.Error", PyExc_StandardError, NULL);
PyDict_SetItemString(dict, "Error", Error);
Warning = PyErr_NewException("sqlite3.Warning", PyExc_StandardError, NULL);
PyDict_SetItemString(dict, "Warning", Warning);
/* Error subclasses */
InterfaceError = PyErr_NewException("sqlite3.InterfaceError", Error, NULL);
PyDict_SetItemString(dict, "InterfaceError", InterfaceError);
DatabaseError = PyErr_NewException("sqlite3.DatabaseError", Error, NULL);
PyDict_SetItemString(dict, "DatabaseError", DatabaseError);
/* DatabaseError subclasses */
InternalError = PyErr_NewException("sqlite3.InternalError", DatabaseError, NULL);
PyDict_SetItemString(dict, "InternalError", InternalError);
OperationalError = PyErr_NewException("sqlite3.OperationalError", DatabaseError, NULL);
PyDict_SetItemString(dict, "OperationalError", OperationalError);
ProgrammingError = PyErr_NewException("sqlite3.ProgrammingError", DatabaseError, NULL);
PyDict_SetItemString(dict, "ProgrammingError", ProgrammingError);
IntegrityError = PyErr_NewException("sqlite3.IntegrityError", DatabaseError,NULL);
PyDict_SetItemString(dict, "IntegrityError", IntegrityError);
DataError = PyErr_NewException("sqlite3.DataError", DatabaseError, NULL);
PyDict_SetItemString(dict, "DataError", DataError);
NotSupportedError = PyErr_NewException("sqlite3.NotSupportedError", DatabaseError, NULL);
PyDict_SetItemString(dict, "NotSupportedError", NotSupportedError);
Py_INCREF((PyObject*)&PyCell_Type);
OptimizedUnicode = (PyObject*)&PyCell_Type;
PyDict_SetItemString(dict, "OptimizedUnicode", OptimizedUnicode);
PyDict_SetItemString(dict, "PARSE_DECLTYPES", PyInt_FromLong(PARSE_DECLTYPES));
PyDict_SetItemString(dict, "PARSE_COLNAMES", PyInt_FromLong(PARSE_COLNAMES));
PyDict_SetItemString(dict, "version", PyString_FromString(PYSQLITE_VERSION));
PyDict_SetItemString(dict, "sqlite_version", PyString_FromString(sqlite3_libversion()));
/* initialize microprotocols layer */
microprotocols_init(dict);
/* initialize the default converters */
converters_init(dict);
time_module = PyImport_ImportModule("time");
time_time = PyObject_GetAttrString(time_module, "time");
time_sleep = PyObject_GetAttrString(time_module, "sleep");
/* Original comment form _bsddb.c in the Python core. This is also still
* needed nowadays for Python 2.3/2.4.
*
* PyEval_InitThreads is called here due to a quirk in python 1.5
* - 2.2.1 (at least) according to Russell Williamson <merel@wt.net>:
* The global interepreter lock is not initialized until the first
* thread is created using thread.start_new_thread() or fork() is
* called. that would cause the ALLOW_THREADS here to segfault due
* to a null pointer reference if no threads or child processes
* have been created. This works around that and is a no-op if
* threads have already been initialized.
* (see pybsddb-users mailing list post on 2002-08-07)
*/
PyEval_InitThreads();
error:
if (PyErr_Occurred())
{
PyErr_SetString(PyExc_ImportError, "_sqlite3: init failed");
}
}

53
Modules/_sqlite/module.h Normal file
View File

@ -0,0 +1,53 @@
/* module.h - definitions for the module
*
* Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_MODULE_H
#define PYSQLITE_MODULE_H
#include "Python.h"
extern PyObject* Error;
extern PyObject* Warning;
extern PyObject* InterfaceError;
extern PyObject* DatabaseError;
extern PyObject* InternalError;
extern PyObject* OperationalError;
extern PyObject* ProgrammingError;
extern PyObject* IntegrityError;
extern PyObject* DataError;
extern PyObject* NotSupportedError;
extern PyObject* OptimizedUnicode;
/* the functions time.time() and time.sleep() */
extern PyObject* time_time;
extern PyObject* time_sleep;
/* A dictionary, mapping colum types (INTEGER, VARCHAR, etc.) to converter
* functions, that convert the SQL value to the appropriate Python value.
* The key is uppercase.
*/
extern PyObject* converters;
#define PARSE_DECLTYPES 1
#define PARSE_COLNAMES 2
#endif

84
Modules/_sqlite/prepare_protocol.c Normal file
View File

@ -0,0 +1,84 @@
/* prepare_protocol.c - the protocol for preparing values for SQLite
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "prepare_protocol.h"
int prepare_protocol_init(SQLitePrepareProtocol* self, PyObject* args, PyObject* kwargs)
{
return 0;
}
void prepare_protocol_dealloc(SQLitePrepareProtocol* self)
{
self->ob_type->tp_free((PyObject*)self);
}
PyTypeObject SQLitePrepareProtocolType= {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"pysqlite2.dbapi2.PrepareProtocol", /* tp_name */
sizeof(SQLitePrepareProtocol), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)prepare_protocol_dealloc, /* tp_dealloc */
0, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
0, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
0, /* tp_hash */
0, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT, /* tp_flags */
0, /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
0, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)prepare_protocol_init, /* tp_init */
0, /* tp_alloc */
0, /* tp_new */
0 /* tp_free */
};
extern int prepare_protocol_setup_types(void)
{
SQLitePrepareProtocolType.tp_new = PyType_GenericNew;
SQLitePrepareProtocolType.ob_type= &PyType_Type;
return PyType_Ready(&SQLitePrepareProtocolType);
}

41
Modules/_sqlite/prepare_protocol.h Normal file
View File

@ -0,0 +1,41 @@
/* prepare_protocol.h - the protocol for preparing values for SQLite
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_PREPARE_PROTOCOL_H
#define PYSQLITE_PREPARE_PROTOCOL_H
#include "Python.h"
typedef struct
{
PyObject_HEAD
} SQLitePrepareProtocol;
extern PyTypeObject SQLitePrepareProtocolType;
int prepare_protocol_init(SQLitePrepareProtocol* self, PyObject* args, PyObject* kwargs);
void prepare_protocol_dealloc(SQLitePrepareProtocol* self);
int prepare_protocol_setup_types(void);
#define UNKNOWN (-1)
#endif

195
Modules/_sqlite/row.c Normal file
View File

@ -0,0 +1,195 @@
/* row.c - an enhanced tuple for database rows
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "row.h"
#include "cursor.h"
void row_dealloc(Row* self)
{
Py_XDECREF(self->data);
Py_XDECREF(self->description);
self->ob_type->tp_free((PyObject*)self);
}
int row_init(Row* self, PyObject* args, PyObject* kwargs)
{
PyObject* data;
Cursor* cursor;
self->data = 0;
self->description = 0;
if (!PyArg_ParseTuple(args, "OO", &cursor, &data)) {
return -1;
}
if (!PyObject_IsInstance((PyObject*)cursor, (PyObject*)&CursorType)) {
PyErr_SetString(PyExc_TypeError, "instance of cursor required for first argument");
return -1;
}
if (!PyTuple_Check(data)) {
PyErr_SetString(PyExc_TypeError, "tuple required for second argument");
return -1;
}
Py_INCREF(data);
self->data = data;
Py_INCREF(cursor->description);
self->description = cursor->description;
return 0;
}
PyObject* row_subscript(Row* self, PyObject* idx)
{
long _idx;
char* key;
int nitems, i;
char* compare_key;
char* p1;
char* p2;
PyObject* item;
if (PyInt_Check(idx)) {
_idx = PyInt_AsLong(idx);
item = PyTuple_GetItem(self->data, _idx);
if (item) {
Py_INCREF(item);
}
return item;
} else if (PyString_Check(idx)) {
key = PyString_AsString(idx);
nitems = PyTuple_Size(self->description);
for (i = 0; i < nitems; i++) {
compare_key = PyString_AsString(PyTuple_GET_ITEM(PyTuple_GET_ITEM(self->description, i), 0));
p1 = key;
p2 = compare_key;
while (1) {
if ((*p1 == (char)0) || (*p2 == (char)0)) {
break;
}
if ((*p1 | 0x20) != (*p2 | 0x20)) {
break;
}
p1++;
p2++;
}
if ((*p1 == (char)0) && (*p2 == (char)0)) {
/* found item */
item = PyTuple_GetItem(self->data, i);
Py_INCREF(item);
return item;
}
}
PyErr_SetString(PyExc_IndexError, "No item with that key");
return NULL;
} else if (PySlice_Check(idx)) {
PyErr_SetString(PyExc_ValueError, "slices not implemented, yet");
return NULL;
} else {
PyErr_SetString(PyExc_IndexError, "Index must be int or string");
return NULL;
}
}
int row_length(Row* self, PyObject* args, PyObject* kwargs)
{
return PyTuple_GET_SIZE(self->data);
}
static int row_print(Row* self, FILE *fp, int flags)
{
return (&PyTuple_Type)->tp_print(self->data, fp, flags);
}
PyMappingMethods row_as_mapping = {
/* mp_length */ (inquiry)row_length,
/* mp_subscript */ (binaryfunc)row_subscript,
/* mp_ass_subscript */ (objobjargproc)0,
};
PyTypeObject RowType = {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"pysqlite2.dbapi2.Row", /* tp_name */
sizeof(Row), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)row_dealloc, /* tp_dealloc */
(printfunc)row_print, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
0, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
0, /* tp_hash */
0, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, /* tp_flags */
0, /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
0, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)row_init, /* tp_init */
0, /* tp_alloc */
0, /* tp_new */
0 /* tp_free */
};
extern int row_setup_types(void)
{
RowType.tp_new = PyType_GenericNew;
RowType.tp_as_mapping = &row_as_mapping;
return PyType_Ready(&RowType);
}

39
Modules/_sqlite/row.h Normal file
View File

@ -0,0 +1,39 @@
/* row.h - an enhanced tuple for database rows
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_ROW_H
#define PYSQLITE_ROW_H
#include "Python.h"
typedef struct _Row
{
PyObject_HEAD
PyObject* data;
PyObject* description;
} Row;
extern PyTypeObject RowType;
int row_setup_types(void);
#endif

285
Modules/_sqlite/statement.c Normal file
View File

@ -0,0 +1,285 @@
/* statement.c - the statement type
*
* Copyright (C) 2005-2006 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "statement.h"
#include "connection.h"
/* prototypes */
int check_remaining_sql(const char* tail);
typedef enum {
LINECOMMENT_1,
IN_LINECOMMENT,
COMMENTSTART_1,
IN_COMMENT,
COMMENTEND_1,
NORMAL
} parse_remaining_sql_state;
int statement_create(Statement* self, Connection* connection, PyObject* sql)
{
const char* tail;
int rc;
PyObject* sql_str;
char* sql_cstr;
self->st = NULL;
self->st = NULL;
self->in_use = 0;
if (PyString_Check(sql)) {
sql_str = sql;
Py_INCREF(sql_str);
} else if (PyUnicode_Check(sql)) {
sql_str = PyUnicode_AsUTF8String(sql);
if (!sql_str) {
rc = PYSQLITE_SQL_WRONG_TYPE;
return rc;
}
} else {
rc = PYSQLITE_SQL_WRONG_TYPE;
return rc;
}
self->sql = sql_str;
sql_cstr = PyString_AsString(sql_str);
rc = sqlite3_prepare(connection->db,
sql_cstr,
-1,
&self->st,
&tail);
self->db = connection->db;
if (rc == SQLITE_OK && check_remaining_sql(tail)) {
(void)sqlite3_finalize(self->st);
rc = PYSQLITE_TOO_MUCH_SQL;
}
return rc;
}
int statement_recompile(Statement* self)
{
const char* tail;
int rc;
char* sql_cstr;
sqlite3_stmt* new_st;
sql_cstr = PyString_AsString(self->sql);
rc = sqlite3_prepare(self->db,
sql_cstr,
-1,
&new_st,
&tail);
if (rc == SQLITE_OK) {
(void)sqlite3_transfer_bindings(self->st, new_st);
(void)sqlite3_finalize(self->st);
self->st = new_st;
}
return rc;
}
int statement_finalize(Statement* self)
{
int rc;
rc = SQLITE_OK;
if (self->st) {
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_finalize(self->st);
Py_END_ALLOW_THREADS
self->st = NULL;
}
self->in_use = 0;
return rc;
}
int statement_reset(Statement* self)
{
int rc;
rc = SQLITE_OK;
if (self->in_use && self->st) {
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_reset(self->st);
Py_END_ALLOW_THREADS
if (rc == SQLITE_OK) {
self->in_use = 0;
}
}
return rc;
}
void statement_mark_dirty(Statement* self)
{
self->in_use = 1;
}
void statement_dealloc(Statement* self)
{
int rc;
if (self->st) {
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_finalize(self->st);
Py_END_ALLOW_THREADS
}
self->st = NULL;
Py_XDECREF(self->sql);
self->ob_type->tp_free((PyObject*)self);
}
/*
* Checks if there is anything left in an SQL string after SQLite compiled it.
* This is used to check if somebody tried to execute more than one SQL command
* with one execute()/executemany() command, which the DB-API and we don't
* allow.
*
* Returns 1 if there is more left than should be. 0 if ok.
*/
int check_remaining_sql(const char* tail)
{
const char* pos = tail;
parse_remaining_sql_state state = NORMAL;
for (;;) {
switch (*pos) {
case 0:
return 0;
case '-':
if (state == NORMAL) {
state = LINECOMMENT_1;
} else if (state == LINECOMMENT_1) {
state = IN_LINECOMMENT;
}
break;
case ' ':
case '\t':
break;
case '\n':
case 13:
if (state == IN_LINECOMMENT) {
state = NORMAL;
}
break;
case '/':
if (state == NORMAL) {
state = COMMENTSTART_1;
} else if (state == COMMENTEND_1) {
state = NORMAL;
} else if (state == COMMENTSTART_1) {
return 1;
}
break;
case '*':
if (state == NORMAL) {
return 1;
} else if (state == LINECOMMENT_1) {
return 1;
} else if (state == COMMENTSTART_1) {
state = IN_COMMENT;
} else if (state == IN_COMMENT) {
state = COMMENTEND_1;
}
break;
default:
if (state == COMMENTEND_1) {
state = IN_COMMENT;
} else if (state == IN_LINECOMMENT) {
} else if (state == IN_COMMENT) {
} else {
return 1;
}
}
pos++;
}
return 0;
}
PyTypeObject StatementType = {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"pysqlite2.dbapi2.Statement", /* tp_name */
sizeof(Statement), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)statement_dealloc, /* tp_dealloc */
0, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
0, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
0, /* tp_hash */
0, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT, /* tp_flags */
0, /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
0, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)0, /* tp_init */
0, /* tp_alloc */
0, /* tp_new */
0 /* tp_free */
};
extern int statement_setup_types(void)
{
StatementType.tp_new = PyType_GenericNew;
return PyType_Ready(&StatementType);
}

55
Modules/_sqlite/statement.h Normal file
View File

@ -0,0 +1,55 @@
/* statement.h - definitions for the statement type
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_STATEMENT_H
#define PYSQLITE_STATEMENT_H
#include "Python.h"
#include "connection.h"
#include "sqlite3.h"
#define PYSQLITE_TOO_MUCH_SQL (-100)
#define PYSQLITE_SQL_WRONG_TYPE (-101)
typedef struct
{
PyObject_HEAD
sqlite3* db;
sqlite3_stmt* st;
PyObject* sql;
int in_use;
} Statement;
extern PyTypeObject StatementType;
int statement_create(Statement* self, Connection* connection, PyObject* sql);
void statement_dealloc(Statement* self);
int statement_recompile(Statement* self);
int statement_finalize(Statement* self);
int statement_reset(Statement* self);
void statement_mark_dirty(Statement* self);
int statement_setup_types(void);
#endif

147
Modules/_sqlite/util.c Normal file
View File

@ -0,0 +1,147 @@
/* util.c - various utility functions
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#include "module.h"
#include "connection.h"
/*
* it's not so trivial to write a portable sleep in C. For now, the simplest
* solution is to just use Python's sleep().
*/
void pysqlite_sleep(double seconds)
{
PyObject* ret;
ret = PyObject_CallFunction(time_sleep, "d", seconds);
Py_DECREF(ret);
}
double pysqlite_time(void)
{
PyObject* ret;
double time;
ret = PyObject_CallFunction(time_time, "");
time = PyFloat_AsDouble(ret);
Py_DECREF(ret);
return time;
}
int _sqlite_step_with_busyhandler(sqlite3_stmt* statement, Connection* connection
)
{
int rc;
Py_BEGIN_ALLOW_THREADS
rc = sqlite3_step(statement);
Py_END_ALLOW_THREADS
return rc;
}
/**
* Checks the SQLite error code and sets the appropriate DB-API exception.
* Returns the error code (0 means no error occured).
*/
int _seterror(sqlite3* db)
{
int errorcode;
errorcode = sqlite3_errcode(db);
switch (errorcode)
{
case SQLITE_OK:
PyErr_Clear();
break;
case SQLITE_ERROR:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_INTERNAL:
PyErr_SetString(InternalError, sqlite3_errmsg(db));
break;
case SQLITE_PERM:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_ABORT:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_BUSY:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_LOCKED:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_NOMEM:
(void)PyErr_NoMemory();
break;
case SQLITE_READONLY:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_INTERRUPT:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_IOERR:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_CORRUPT:
PyErr_SetString(DatabaseError, sqlite3_errmsg(db));
break;
case SQLITE_NOTFOUND:
PyErr_SetString(InternalError, sqlite3_errmsg(db));
break;
case SQLITE_FULL:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_CANTOPEN:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_PROTOCOL:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_EMPTY:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_SCHEMA:
PyErr_SetString(OperationalError, sqlite3_errmsg(db));
break;
case SQLITE_TOOBIG:
PyErr_SetString(DataError, sqlite3_errmsg(db));
break;
case SQLITE_CONSTRAINT:
PyErr_SetString(IntegrityError, sqlite3_errmsg(db));
break;
case SQLITE_MISMATCH:
PyErr_SetString(IntegrityError, sqlite3_errmsg(db));
break;
case SQLITE_MISUSE:
PyErr_SetString(ProgrammingError, sqlite3_errmsg(db));
break;
default:
PyErr_SetString(DatabaseError, sqlite3_errmsg(db));
}
return errorcode;
}

40
Modules/_sqlite/util.h Normal file
View File

@ -0,0 +1,40 @@
/* util.h - various utility functions
*
* Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
* This software is provided 'as-is', without any express or implied
* warranty. In no event will the authors be held liable for any damages
* arising from the use of this software.
*
* Permission is granted to anyone to use this software for any purpose,
* including commercial applications, and to alter it and redistribute it
* freely, subject to the following restrictions:
*
* 1. The origin of this software must not be misrepresented; you must not
* claim that you wrote the original software. If you use this software
* in a product, an acknowledgment in the product documentation would be
* appreciated but is not required.
* 2. Altered source versions must be plainly marked as such, and must not be
* misrepresented as being the original software.
* 3. This notice may not be removed or altered from any source distribution.
*/
#ifndef PYSQLITE_UTIL_H
#define PYSQLITE_UTIL_H
#include "Python.h"
#include "pythread.h"
#include "sqlite3.h"
#include "connection.h"
void pysqlite_sleep(double seconds);
int _sqlite_step_with_busyhandler(sqlite3_stmt* statement, Connection* connection);
/**
* Checks the SQLite error code and sets the appropriate DB-API exception.
* Returns the error code (0 means no error occured).
*/
int _seterror(sqlite3* db);
#endif

75
setup.py
View File

@ -689,6 +689,81 @@ class PyBuildExt(build_ext):
dblibs = []
dblib_dir = None
# The sqlite interface
# We hunt for "#define SQLITE_VERSION_NUMBER nnnnn"
# We need to find a version >= 3002002 (> sqlite version 3.2.2)
sqlite_incdir = sqlite_libdir = None
sqlite_inc_paths = [ '/usr/include',
'/usr/include/sqlite',
'/usr/include/sqlite3',
'/usr/local/include',
'/usr/local/include/sqlite',
'/usr/local/include/sqlite3',
]
MIN_SQLITE_VERSION = 3002002
for d in sqlite_inc_paths + inc_dirs:
f = os.path.join(d, "sqlite3.h")
if os.path.exists(f):
if db_setup_debug: print "found %s"%f
f = open(f).read()
m = re.search(r"#define\WSQLITE_VERSION_NUMBER\W(\d+)", f)
if m:
sqlite_version = int(m.group(1))
if sqlite_version >= MIN_SQLITE_VERSION:
# we win!
print "%s/sqlite3.h: version %s"%(d, sqlite_version)
sqlite_incdir = d
break
else:
if db_setup_debug:
print "%s: version %d is too old, need >= %s"%(d,
sqlite_version, MIN_SQLITE_VERSION)
if sqlite_incdir:
sqlite_dirs_to_check = [
os.path.join(sqlite_incdir, '..', 'lib64'),
os.path.join(sqlite_incdir, '..', 'lib'),
os.path.join(sqlite_incdir, '..', '..', 'lib64'),
os.path.join(sqlite_incdir, '..', '..', 'lib'),
]
sqlite_libfile = self.compiler.find_library_file(
sqlite_dirs_to_check + lib_dirs, 'sqlite3')
sqlite_libdir = [os.path.abspath(os.path.dirname(sqlite_libfile))]
if sqlite_incdir and sqlite_libdir:
sqlite_srcs = ['_sqlite/adapters.c',
'_sqlite/cache.c',
'_sqlite/connection.c',
'_sqlite/converters.c',
'_sqlite/cursor.c',
'_sqlite/microprotocols.c',
'_sqlite/module.c',
'_sqlite/prepare_protocol.c',
'_sqlite/row.c',
'_sqlite/statement.c',
'_sqlite/util.c', ]
PYSQLITE_VERSION = "2.1.3"
sqlite_defines = []
if sys.platform != "win32":
sqlite_defines.append(('PYSQLITE_VERSION',
'"%s"' % PYSQLITE_VERSION))
else:
sqlite_defines.append(('PYSQLITE_VERSION',
'\\"'+PYSQLITE_VERSION+'\\"'))
sqlite_defines.append(('PY_MAJOR_VERSION',
str(sys.version_info[0])))
sqlite_defines.append(('PY_MINOR_VERSION',
str(sys.version_info[1])))
exts.append(Extension('_sqlite3', sqlite_srcs,
define_macros=sqlite_defines,
include_dirs=["Modules/_sqlite",
sqlite_incdir],
library_dirs=sqlite_libdir,
runtime_library_dirs=sqlite_libdir,
libraries=["sqlite3",]))
# Look for Berkeley db 1.85. Note that it is built as a different
# module name so it can be included even when later versions are