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