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  connection = None 
 39   
 40   
41 -class NoInsertionError(Exception):
42 pass
43 44
45 -def _check_is_public(name):
46 if name[:6] == "_names": 47 raise AttributeError
48 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, AttributeError): 69 raise AttributeError("'%s' object has no attribute '%s'" 70 % (self.__class__.__name__, name))
71
72 - def __setattr__(self, name, value):
73 try: 74 self._names_hash 75 except AttributeError: 76 return object.__setattr__(self, name, value) 77 78 _check_is_public(name) 79 try: 80 index = self._names_hash[name] 81 self[index] = value 82 except KeyError: 83 return object.__setattr__(self, name, value)
84 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
125 -class QueryGeneric(Query):
126 - def __init__(self, statement, *args, **keywds):
127 Query.__init__(self, *args, **keywds) 128 self.statement = statement,
129 130
131 -class IterationCursor(object):
132 - def __init__(self, query, connection=connection):
133 if connection is None: 134 raise TypeError("database connection is None") 135 self.cursor = connection.cursor() 136 self.row_class = query.row_class 137 if query.diagnostics: 138 sys.stderr.write("Query statement: %s\n" % query.statement) 139 sys.stderr.write("Query params: %s\n" % query.params) 140 self.cursor.execute(query.statement, query.params)
141
142 - def __next__(self):
143 return self.row_class(self.cursor)
144 145 if sys.version_info[0] < 3:
146 - def next(self):
147 """Deprecated Python 2 style alias for Python 3 style __next__ method.""" 148 import warnings 149 from Bio import BiopythonDeprecationWarning 150 warnings.warn("Please use next(my_iterator) instead of my_iterator.next(), " 151 "the .next() method is deprecated and will be removed in a " 152 "future release of Biopython.", BiopythonDeprecationWarning) 153 return self.__next__()
154 155
156 -class QuerySingle(Query, QueryRow):
157 ignore_warnings = 0 158
159 - def __init__(self, *args, **keywds):
160 message = self.MSG_FAILURE 161 Query.__init__(self, *args, **keywds) 162 try: 163 self.single_cursor = Query.cursor(self) 164 except MySQLdb.Warning: 165 if not self.ignore_warnings: 166 raise 167 self.row_class.__init__(self, self.cursor()) 168 object.__setattr__(self, "message", self.MSG_SUCCESS)
169
170 - def cursor(self):
171 return self.single_cursor
172 173
174 -class QueryAll(list, Query):
175 - def __init__(self, *args, **keywds):
176 Query.__init__(self, *args, **keywds) 177 list.__init__(self, [self.process_row(r) for r in self.cursor().fetchall()])
178
179 - def process_row(self, row):
180 return row
181 182
183 -class QueryAllFirstItem(QueryAll):
184 - def process_row(self, row):
185 return row[0]
186 187
188 -class Create(QuerySingle):
189 - def __init__(self, *args, **keywds):
190 try: 191 QuerySingle.__init__(self, *args, **keywds) 192 except StopIteration: 193 self.message = self.MSG_SUCCESS
194 195
196 -class Update(Create):
197 pass
198 199
200 -class Insert(Create):
201 MSG_INTEGRITY_ERROR = "Couldn't insert: %s. " 202
203 - def __init__(self, *args, **keywds):
204 try: 205 Create.__init__(self, *args, **keywds) 206 except MySQLdb.IntegrityError as error_data: 207 self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1] 208 try: 209 self.total_count 210 except AttributeError: 211 self.total_count = 0 212 213 raise MySQLdb.IntegrityError(self.error_message) 214 215 self.id = self.cursor().insert_id() 216 try: 217 self.total_count += self.cursor().rowcount 218 except AttributeError: 219 self.total_count = self.cursor().rowcount 220 221 if self.cursor().rowcount == 0: 222 raise NoInsertionError
223 224
225 -def _test(*args, **keywds):
226 import doctest 227 doctest.testmod(sys.modules[__name__], *args, **keywds)
228 229 if __name__ == "__main__": 230 if __debug__: 231 _test() 232