1
2
3
4
5
6
7
8
9
10
11 _dbutils = {}
12
14 """Default database utilities."""
17
19 if table != 'biosequence': return table
20 else: return 'bioentry'
21
23
24 table = self.tname(table)
25 sql = r"select max(%s_id) from %s" % (table, table)
26 cursor.execute(sql)
27 rv = cursor.fetchone()
28 return rv[0]
29
30 - def execute(self, cursor, sql, args=None):
31 """Just execute an sql command.
32 """
33 cursor.execute(sql, args or ())
34
38
39
41 """Custom database utilities for SQLite."""
42 - def execute(self, cursor, sql, args=None):
43 """Execute SQL command, replacing %s with ? for variable substitution in sqlite3.
44 """
45 cursor.execute(sql.replace("%s", "?"), args or ())
46
47 _dbutils["sqlite3"] = Sqlite_dbutils
48
49
51 """Custom database utilities for MySQL."""
53 try:
54
55 return cursor.insert_id()
56 except AttributeError:
57
58
59
60 return cursor.lastrowid
61
62 _dbutils["MySQLdb"] = Mysql_dbutils
63
64
65 -class _PostgreSQL_dbutils(Generic_dbutils):
66 """Base class for any PostgreSQL adaptor."""
67 - def next_id(self, cursor, table):
68 table = self.tname(table)
69 sql = r"select nextval('%s_pk_seq')" % table
70 cursor.execute(sql)
71 rv = cursor.fetchone()
72 return rv[0]
73
74 - def last_id(self, cursor, table):
75 table = self.tname(table)
76 sql = r"select currval('%s_pk_seq')" % table
77 cursor.execute(sql)
78 rv = cursor.fetchone()
79 return rv[0]
80
82 """Custom database utilities for Psycopg (PostgreSQL)."""
85
86 _dbutils["psycopg"] = Psycopg_dbutils
87
88
90 """Custom database utilities for Psycopg2 (PostgreSQL)."""
92 if y:
93 conn.set_isolation_level(0)
94 else:
95 conn.set_isolation_level(1)
96
97 _dbutils["psycopg2"] = Psycopg2_dbutils
98
99
101 """Custom database utilities for Pgdb (aka PyGreSQL, for PostgreSQL)."""
103 raise NotImplementedError("pgdb does not support this!")
104
105 _dbutils["pgdb"] = Pgdb_dbutils
106
107
113