Package BioSQL :: Module DBUtils
[hide private]
[frames] | no frames]

Source Code for Module BioSQL.DBUtils

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