Bio.DocSQL

1 #!/usr/bin/env python 2 # 3 # Copyright 2002-2003 by Michael Hoffman. All rights reserved. 4 # This code is part of the Biopython distribution and governed by its 5 # license. Please see the LICENSE file that should have been included 6 # as part of this package. 7 8 # make yield compatible with Python2.2 9 from __future__ import generators 10 11 """ 12 Bio.DocSQL: easy access to DB API databases. 13 14 >>> import DocSQL, MySQLdb, os 15 >>> db=MySQLdb.connect(passwd='', db='test') 16 >>> class CreatePeople(DocSQL.Create): 17 ... \""" 18 ... CREATE TEMPORARY TABLE people 19 ... (id INT UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT, 20 ... last_name TINYTEXT, 21 ... first_name TINYTEXT) 22 ... \""" 23 ... 24 >>> CreatePeople(connection=db) 25 CreatePeople(message=Success) 26 """ 27 28 __version__ = "$Revision: 1.11 $" 29 # $Source: /home/repository/biopython/biopython/Bio/DocSQL.py,v $ 30 31 import exceptions 32 import sys 33 34 from Bio import MissingExternalDependencyError 35 36 try: 37 import MySQLdb 38 except: 39 raise MissingExternalDependencyError("Install MySQLdb if you want to use Bio.DocSQL.") 40 41 connection = None 42

43 -class NoInsertionError(exceptions.Exception):

44 pass

45

46 -def _check_is_public(name):

47 if name[:6] == "_names": 48 raise AttributeError

49

50 -class QueryRow(list):

51 - def __init__(self, cursor):

52 try: 53 row = cursor.fetchone() 54 super(QueryRow, self).__init__(row) 55 except TypeError: 56 raise StopIteration 57 58 object.__setattr__(self, "_names", [x[0] for x in cursor.description]) # FIXME: legacy 59 object.__setattr__(self, "_names_hash", {}) 60 61 for i, name in enumerate(self._names): 62 self._names_hash[name] = i

63

64 - def __getattr__(self, name):

65 _check_is_public(name) 66 try: 67 return self[self._names_hash[name]] 68 except KeyError: 69 raise AttributeError, "'%s' object has no attribute '%s'" % (self.__class__.__name__, name) 70 except AttributeError: 71 raise AttributeError, "'%s' object has no attribute '%s'" % (self.__class__.__name__, name)

72

73 - def __setattr__(self, name, value):

74 try: 75 self._names_hash 76 except AttributeError: 77 return object.__setattr__(self, name, value) 78 79 _check_is_public(name) 80 try: 81 index = self._names_hash[name] 82 self[index] = value 83 except KeyError: 84 return object.__setattr__(self, name, value)

85

86 -class Query(object):

87 """ 88 SHOW TABLES 89 """ 90 MSG_FAILURE = "Failure" 91 MSG_SUCCESS = "Success" 92 message = "not executed" 93 error_message = "" 94 prefix = "" 95 suffix = "" 96 row_class = QueryRow 97

98 - def __init__(self, *args, **keywds):

99 try: 100 self.connection = keywds['connection'] 101 except KeyError: 102 self.connection = connection 103 try: 104 self.diagnostics = keywds['diagnostics'] 105 except KeyError: 106 self.diagnostics = 0 107 108 self.statement = self.prefix + self.__doc__ + self.suffix 109 self.params = args

110

111 - def __iter__(self):

112 return IterationCursor(self, self.connection)

113

114 - def __repr__(self):

115 return "%s(message=%s)" % (self.__class__.__name__, self.message)

116

117 - def cursor(self):

118 return iter(self).cursor

119

120 - def dump(self):

121 for item in self: 122 print item

123

124 -class QueryGeneric(Query):

125 - def __init__(self, statement, *args, **keywds):

126 Query.__init__(self, *args, **keywds) 127 self.statement = statement,

128

129 -class IterationCursor(object):

130 - def __init__(self, query, connection=connection):

131 if connection is None: 132 raise TypeError, "database connection is None" 133 self.cursor = connection.cursor() 134 self.row_class = query.row_class 135 if query.diagnostics: 136 print >>sys.stderr, query.statement 137 print >>sys.stderr, query.params 138 self.cursor.execute(query.statement, query.params)

139

140 - def next(self):

141 return self.row_class(self.cursor)

142

143 -class QuerySingle(Query, QueryRow):

144 ignore_warnings = 0

145 - def __init__(self, *args, **keywds):

146 message = self.MSG_FAILURE 147 Query.__init__(self, *args, **keywds) 148 try: 149 self.single_cursor = Query.cursor(self) 150 except MySQLdb.Warning: 151 if not self.ignore_warnings: 152 raise 153 self.row_class.__init__(self, self.cursor()) 154 object.__setattr__(self, "message", self.MSG_SUCCESS)

155

156 - def cursor(self):

157 return self.single_cursor

158

159 -class QueryAll(list, Query):

160 - def __init__(self, *args, **keywds):

161 Query.__init__(self, *args, **keywds) 162 list.__init__(self, map(self.process_row, self.cursor().fetchall()))

163

164 - def process_row(self, row):

165 return row

166

167 -class QueryAllFirstItem(QueryAll):

168 - def process_row(self, row):

169 return row[0]

170

171 -class Create(QuerySingle):

172 - def __init__(self, *args, **keywds):

173 try: 174 QuerySingle.__init__(self, *args, **keywds) 175 except StopIteration: 176 self.message = self.MSG_SUCCESS

177

178 -class Update(Create):

179 pass

180

181 -class Insert(Create):

182 MSG_INTEGRITY_ERROR = "Couldn't insert: %s. " 183

184 - def __init__(self, *args, **keywds):

185 try: 186 Create.__init__(self, *args, **keywds) 187 except MySQLdb.IntegrityError, error_data: 188 self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1] 189 try: 190 self.total_count 191 except AttributeError: 192 self.total_count = 0 193 194 raise MySQLdb.IntegrityError, self.error_message 195 196 self.id = self.cursor().insert_id() 197 try: 198 self.total_count += self.cursor().rowcount 199 except AttributeError: 200 self.total_count = self.cursor().rowcount 201 202 if self.cursor().rowcount == 0: 203 raise NoInsertionError

204

205 -def _test(*args, **keywds):

206 import doctest, sys 207 doctest.testmod(sys.modules[__name__], *args, **keywds)

208 209 if __name__ == "__main__": 210 if __debug__: 211 _test() 212

Source Code for Module Bio.DocSQL