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