Package Bio :: Module DocSQL
[hide private]
[frames] | no frames]

Source Code for Module 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