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