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.
9 Bio.DocSQL: easy access to DB API databases.
11 >>> import DocSQL, MySQLdb, os
12 >>> db=MySQLdb.connect(passwd='', db='test')
13 >>> class CreatePeople(DocSQL.Create):
15 ... CREATE TEMPORARY TABLE people
16 ... (id INT UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT,
17 ... last_name TINYTEXT,
18 ... first_name TINYTEXT)
21 >>> CreatePeople(connection=db)
22 CreatePeople(message=Success)
25 __version__ = "$Revision: 1.13 $"
26 # $Source: /home/repository/biopython/biopython/Bio/DocSQL.py,v $
31 from Bio import MissingExternalDependencyError
36 raise MissingExternalDependencyError("Install MySQLdb if you want to use Bio.DocSQL.")
40 class NoInsertionError(exceptions.Exception):
43 def _check_is_public(name):
44 if name[:6] == "_names":
48 def __init__(self, cursor):
50 row = cursor.fetchone()
51 super(QueryRow, self).__init__(row)
55 object.__setattr__(self, "_names", [x[0] for x in cursor.description]) # FIXME: legacy
56 object.__setattr__(self, "_names_hash", {})
58 for i, name in enumerate(self._names):
59 self._names_hash[name] = i
61 def __getattr__(self, name):
62 _check_is_public(name)
64 return self[self._names_hash[name]]
65 except (KeyError, AttributeError) :
66 raise AttributeError("'%s' object has no attribute '%s'" \
67 % (self.__class__.__name__, name))
69 def __setattr__(self, name, value):
72 except AttributeError:
73 return object.__setattr__(self, name, value)
75 _check_is_public(name)
77 index = self._names_hash[name]
80 return object.__setattr__(self, name, value)
86 MSG_FAILURE = "Failure"
87 MSG_SUCCESS = "Success"
88 message = "not executed"
94 def __init__(self, *args, **keywds):
96 self.connection = keywds['connection']
98 self.connection = connection
100 self.diagnostics = keywds['diagnostics']
104 self.statement = self.prefix + self.__doc__ + self.suffix
108 return IterationCursor(self, self.connection)
111 return "%s(message=%s)" % (self.__class__.__name__, self.message)
114 return iter(self).cursor
120 class QueryGeneric(Query):
121 def __init__(self, statement, *args, **keywds):
122 Query.__init__(self, *args, **keywds)
123 self.statement = statement,
125 class IterationCursor(object):
126 def __init__(self, query, connection=connection):
127 if connection is None:
128 raise TypeError("database connection is None")
129 self.cursor = connection.cursor()
130 self.row_class = query.row_class
131 if query.diagnostics:
132 print >>sys.stderr, query.statement
133 print >>sys.stderr, query.params
134 self.cursor.execute(query.statement, query.params)
137 return self.row_class(self.cursor)
139 class QuerySingle(Query, QueryRow):
141 def __init__(self, *args, **keywds):
142 message = self.MSG_FAILURE
143 Query.__init__(self, *args, **keywds)
145 self.single_cursor = Query.cursor(self)
146 except MySQLdb.Warning:
147 if not self.ignore_warnings:
149 self.row_class.__init__(self, self.cursor())
150 object.__setattr__(self, "message", self.MSG_SUCCESS)
153 return self.single_cursor
155 class QueryAll(list, Query):
156 def __init__(self, *args, **keywds):
157 Query.__init__(self, *args, **keywds)
158 list.__init__(self, map(self.process_row, self.cursor().fetchall()))
160 def process_row(self, row):
163 class QueryAllFirstItem(QueryAll):
164 def process_row(self, row):
167 class Create(QuerySingle):
168 def __init__(self, *args, **keywds):
170 QuerySingle.__init__(self, *args, **keywds)
171 except StopIteration:
172 self.message = self.MSG_SUCCESS
174 class Update(Create):
177 class Insert(Create):
178 MSG_INTEGRITY_ERROR = "Couldn't insert: %s. "
180 def __init__(self, *args, **keywds):
182 Create.__init__(self, *args, **keywds)
183 except MySQLdb.IntegrityError, error_data:
184 self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1]
187 except AttributeError:
190 raise MySQLdb.IntegrityError(self.error_message)
192 self.id = self.cursor().insert_id()
194 self.total_count += self.cursor().rowcount
195 except AttributeError:
196 self.total_count = self.cursor().rowcount
198 if self.cursor().rowcount == 0:
199 raise NoInsertionError
201 def _test(*args, **keywds):
203 doctest.testmod(sys.modules[__name__], *args, **keywds)
205 if __name__ == "__main__":