1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
# Copyright 2002 by Andrew Dalke. All rights reserved.
# Revisions 2007-2010 copyright by Peter Cock. All rights reserved.
# Revisions 2009 copyright by Brad Chapman. All rights reserved.
# Revisions 2013 copyright by Tiago Antao. All rights reserved.
# This code is part of the Biopython distribution and governed by its
# license. Please see the LICENSE file that should have been included
# as part of this package.
#
# Note that BioSQL (including the database schema and scripts) is
# available and licensed separately. Please consult www.biosql.org
import os
_dbutils = {}
class Generic_dbutils:
"""Default database utilities."""
def __init__(self):
pass
def tname(self, table):
if table != 'biosequence':
return table
else:
return 'bioentry'
def last_id(self, cursor, table):
# XXX: Unsafe without transactions isolation
table = self.tname(table)
sql = r"select max(%s_id) from %s" % (table, table)
cursor.execute(sql)
rv = cursor.fetchone()
return rv[0]
def execute(self, cursor, sql, args=None):
"""Just execute an sql command.
"""
cursor.execute(sql, args or ())
def autocommit(self, conn, y=1):
# Let's hope it was not really needed
pass
class Sqlite_dbutils(Generic_dbutils):
"""Custom database utilities for SQLite."""
def execute(self, cursor, sql, args=None):
"""Execute SQL command, replacing %s with ? for variable substitution in sqlite3.
"""
cursor.execute(sql.replace("%s", "?"), args or ())
_dbutils["sqlite3"] = Sqlite_dbutils
class Mysql_dbutils(Generic_dbutils):
"""Custom database utilities for MySQL."""
def last_id(self, cursor, table):
if os.name == "java":
return Generic_dbutils.last_id(self, cursor, table)
try:
#This worked on older versions of MySQL
return cursor.insert_id()
except AttributeError:
#See bug 2390
#Google suggests this is the new way,
#same fix also suggested by Eric Gibert:
return cursor.lastrowid
_dbutils["MySQLdb"] = Mysql_dbutils
class _PostgreSQL_dbutils(Generic_dbutils):
"""Base class for any PostgreSQL adaptor."""
def next_id(self, cursor, table):
table = self.tname(table)
sql = r"select nextval('%s_pk_seq')" % table
cursor.execute(sql)
rv = cursor.fetchone()
return rv[0]
def last_id(self, cursor, table):
table = self.tname(table)
sql = r"select currval('%s_pk_seq')" % table
cursor.execute(sql)
rv = cursor.fetchone()
return rv[0]
class Psycopg2_dbutils(_PostgreSQL_dbutils):
"""Custom database utilities for Psycopg2 (PostgreSQL)."""
def autocommit(self, conn, y=True):
if y:
if os.name == "java":
conn.autocommit = 1
else:
conn.set_isolation_level(0)
else:
if os.name == "java":
conn.autocommit = 0
else:
conn.set_isolation_level(1)
_dbutils["psycopg2"] = Psycopg2_dbutils
class Pgdb_dbutils(_PostgreSQL_dbutils):
"""Custom database utilities for Pgdb (aka PyGreSQL, for PostgreSQL)."""
def autocommit(self, conn, y=True):
raise NotImplementedError("pgdb does not support this!")
_dbutils["pgdb"] = Pgdb_dbutils
def get_dbutils(module_name):
try:
return _dbutils[module_name]()
except KeyError:
return Generic_dbutils()
|
