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