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   
 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              print >>sys.stderr, query.statement 
141              print >>sys.stderr, query.params 
142          self.cursor.execute(query.statement, query.params) 
143   
144 -    def next(self): 
145          return self.row_class(self.cursor) 
146   
147   
148 -class QuerySingle(Query, QueryRow): 
149      ignore_warnings = 0 
150   
151 -    def __init__(self, *args, **keywds): 
152          message = self.MSG_FAILURE 
153          Query.__init__(self, *args, **keywds) 
154          try: 
155              self.single_cursor = Query.cursor(self) 
156          except MySQLdb.Warning: 
157              if not self.ignore_warnings: 
158                  raise 
159          self.row_class.__init__(self, self.cursor()) 
160          object.__setattr__(self, "message", self.MSG_SUCCESS) 
161   
162 -    def cursor(self): 
163          return self.single_cursor 
164   
165   
166 -class QueryAll(list, Query): 
167 -    def __init__(self, *args, **keywds): 
168          Query.__init__(self, *args, **keywds) 
169          list.__init__(self, map(self.process_row, self.cursor().fetchall())) 
170   
171 -    def process_row(self, row): 
172          return row 
173   
174   
175 -class QueryAllFirstItem(QueryAll): 
176 -    def process_row(self, row): 
177          return row[0] 
178   
179   
180 -class Create(QuerySingle): 
181 -    def __init__(self, *args, **keywds): 
182          try: 
183              QuerySingle.__init__(self, *args, **keywds) 
184          except StopIteration: 
185              self.message = self.MSG_SUCCESS 
186   
187   
188 -class Update(Create): 
189      pass 
190   
191   
192 -class Insert(Create): 
193      MSG_INTEGRITY_ERROR = "Couldn't insert: %s. " 
194   
195 -    def __init__(self, *args, **keywds): 
196          try: 
197              Create.__init__(self, *args, **keywds) 
198          except MySQLdb.IntegrityError, error_data: 
199              self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1] 
200              try: 
201                  self.total_count 
202              except AttributeError: 
203                  self.total_count = 0 
204   
205              raise MySQLdb.IntegrityError(self.error_message) 
206   
207          self.id = self.cursor().insert_id() 
208          try: 
209              self.total_count += self.cursor().rowcount 
210          except AttributeError: 
211              self.total_count = self.cursor().rowcount 
212   
213          if self.cursor().rowcount == 0: 
214              raise NoInsertionError 
215   
216   
217 -def _test(*args, **keywds): 
218      import doctest 
219      doctest.testmod(sys.modules[__name__], *args, **keywds) 
220   
221  if __name__ == "__main__": 
222      if __debug__: 
223          _test() 
224