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  """ 
  9  Bio.DocSQL: easy access to DB API databases. 
 10   
 11  >>> import DocSQL, MySQLdb, os 
 12  >>> db=MySQLdb.connect(passwd='', db='test') 
 13  >>> class CreatePeople(DocSQL.Create): 
 14  ...     \""" 
 15  ...     CREATE TEMPORARY TABLE people 
 16  ...     (id INT UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT, 
 17  ...     last_name TINYTEXT, 
 18  ...     first_name TINYTEXT) 
 19  ...     \""" 
 20  ... 
 21  >>> CreatePeople(connection=db) 
 22  CreatePeople(message=Success) 
 23  """ 
 24   
 25  __version__ = "$Revision: 1.13 $" 
 26  # $Source: /home/bartek/cvs2bzr/biopython_fastimport/cvs_repo/biopython/Bio/DocSQL.py,v $ 
 27   
 28  import exceptions 
 29  import sys 
 30   
 31  from Bio import MissingExternalDependencyError 
 32   
 33  try: 
 34      import MySQLdb 
 35  except: 
 36      raise MissingExternalDependencyError("Install MySQLdb if you want to use Bio.DocSQL.") 
 37   
 38  connection = None 
 39   
40 -class NoInsertionError(exceptions.Exception):
41 pass
42
43 -def _check_is_public(name):
44 if name[:6] == "_names": 45 raise AttributeError
46
47 -class QueryRow(list):
48 - def __init__(self, cursor):
49 try: 50 row = cursor.fetchone() 51 super(QueryRow, self).__init__(row) 52 except TypeError: 53 raise StopIteration 54 55 object.__setattr__(self, "_names", [x[0] for x in cursor.description]) # FIXME: legacy 56 object.__setattr__(self, "_names_hash", {}) 57 58 for i, name in enumerate(self._names): 59 self._names_hash[name] = i
60
61 - def __getattr__(self, name):
62 _check_is_public(name) 63 try: 64 return self[self._names_hash[name]] 65 except (KeyError, AttributeError): 66 raise AttributeError("'%s' object has no attribute '%s'" \ 67 % (self.__class__.__name__, name))
68
69 - def __setattr__(self, name, value):
70 try: 71 self._names_hash 72 except AttributeError: 73 return object.__setattr__(self, name, value) 74 75 _check_is_public(name) 76 try: 77 index = self._names_hash[name] 78 self[index] = value 79 except KeyError: 80 return object.__setattr__(self, name, value)
81
82 -class Query(object):
83 """ 84 SHOW TABLES 85 """ 86 MSG_FAILURE = "Failure" 87 MSG_SUCCESS = "Success" 88 message = "not executed" 89 error_message = "" 90 prefix = "" 91 suffix = "" 92 row_class = QueryRow 93
94 - def __init__(self, *args, **keywds):
95 try: 96 self.connection = keywds['connection'] 97 except KeyError: 98 self.connection = connection 99 try: 100 self.diagnostics = keywds['diagnostics'] 101 except KeyError: 102 self.diagnostics = 0 103 104 self.statement = self.prefix + self.__doc__ + self.suffix 105 self.params = args
106
107 - def __iter__(self):
108 return IterationCursor(self, self.connection)
109
110 - def __repr__(self):
111 return "%s(message=%s)" % (self.__class__.__name__, self.message)
112
113 - def cursor(self):
114 return iter(self).cursor
115
116 - def dump(self):
117 for item in self: 118 print item
119
120 -class QueryGeneric(Query):
121 - def __init__(self, statement, *args, **keywds):
122 Query.__init__(self, *args, **keywds) 123 self.statement = statement,
124
125 -class IterationCursor(object):
126 - def __init__(self, query, connection=connection):
127 if connection is None: 128 raise TypeError("database connection is None") 129 self.cursor = connection.cursor() 130 self.row_class = query.row_class 131 if query.diagnostics: 132 print >>sys.stderr, query.statement 133 print >>sys.stderr, query.params 134 self.cursor.execute(query.statement, query.params)
135
136 - def next(self):
137 return self.row_class(self.cursor)
138
139 -class QuerySingle(Query, QueryRow):
140 ignore_warnings = 0
141 - def __init__(self, *args, **keywds):
142 message = self.MSG_FAILURE 143 Query.__init__(self, *args, **keywds) 144 try: 145 self.single_cursor = Query.cursor(self) 146 except MySQLdb.Warning: 147 if not self.ignore_warnings: 148 raise 149 self.row_class.__init__(self, self.cursor()) 150 object.__setattr__(self, "message", self.MSG_SUCCESS)
151
152 - def cursor(self):
153 return self.single_cursor
154
155 -class QueryAll(list, Query):
156 - def __init__(self, *args, **keywds):
157 Query.__init__(self, *args, **keywds) 158 list.__init__(self, map(self.process_row, self.cursor().fetchall()))
159
160 - def process_row(self, row):
161 return row
162
163 -class QueryAllFirstItem(QueryAll):
164 - def process_row(self, row):
165 return row[0]
166
167 -class Create(QuerySingle):
168 - def __init__(self, *args, **keywds):
169 try: 170 QuerySingle.__init__(self, *args, **keywds) 171 except StopIteration: 172 self.message = self.MSG_SUCCESS
173
174 -class Update(Create):
175 pass
176
177 -class Insert(Create):
178 MSG_INTEGRITY_ERROR = "Couldn't insert: %s. " 179
180 - def __init__(self, *args, **keywds):
181 try: 182 Create.__init__(self, *args, **keywds) 183 except MySQLdb.IntegrityError, error_data: 184 self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1] 185 try: 186 self.total_count 187 except AttributeError: 188 self.total_count = 0 189 190 raise MySQLdb.IntegrityError(self.error_message) 191 192 self.id = self.cursor().insert_id() 193 try: 194 self.total_count += self.cursor().rowcount 195 except AttributeError: 196 self.total_count = self.cursor().rowcount 197 198 if self.cursor().rowcount == 0: 199 raise NoInsertionError
200
201 -def _test(*args, **keywds):
202 import doctest, sys 203 doctest.testmod(sys.modules[__name__], *args, **keywds)
204 205 if __name__ == "__main__": 206 if __debug__: 207 _test() 208