website/archive/binaries/mac/src/globplot/biopython-1.50/Bio/DocSQL.py

   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 DocSQL, MySQLdb, os
  12 >>> db=MySQLdb.connect(passwd='', db='test')
  13 >>> class CreatePeople(DocSQL.Create):
  14 ...     \"""
  15 ...     CREATE TEMPORARY TABLE people
  16 ...     (id INT UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT,
  17 ...     last_name TINYTEXT,
  18 ...     first_name TINYTEXT)
  19 ...     \"""
  20 ...
  21 >>> CreatePeople(connection=db)
  22 CreatePeople(message=Success)
  23 """
  24
  25 __version__ = "$Revision: 1.13 $"
  26 # $Source: /home/repository/biopython/biopython/Bio/DocSQL.py,v $
  27
  28 import exceptions
  29 import sys
  30
  31 from Bio import MissingExternalDependencyError
  32
  33 try:
  34     import MySQLdb
  35 except:
  36     raise MissingExternalDependencyError("Install MySQLdb if you want to use Bio.DocSQL.")
  37
  38 connection = None
  39
  40 class NoInsertionError(exceptions.Exception):
  41     pass
  42
  43 def _check_is_public(name):
  44     if name[:6] == "_names":
  45         raise AttributeError
  46
  47 class QueryRow(list):
  48     def __init__(self, cursor):
  49         try:
  50             row = cursor.fetchone()
  51             super(QueryRow, self).__init__(row)
  52         except TypeError:
  53             raise StopIteration
  54
  55         object.__setattr__(self, "_names", [x[0] for x in cursor.description]) # FIXME: legacy
  56         object.__setattr__(self, "_names_hash", {})
  57
  58         for i, name in enumerate(self._names):
  59             self._names_hash[name] = i
  60
  61     def __getattr__(self, name):
  62         _check_is_public(name)
  63         try:
  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))
  68
  69     def __setattr__(self, name, value):
  70         try:
  71             self._names_hash
  72         except AttributeError:
  73             return object.__setattr__(self, name, value)
  74
  75         _check_is_public(name)
  76         try:
  77             index = self._names_hash[name]
  78             self[index] = value
  79         except KeyError:
  80             return object.__setattr__(self, name, value)
  81
  82 class Query(object):
  83     """
  84     SHOW TABLES
  85     """
  86     MSG_FAILURE = "Failure"
  87     MSG_SUCCESS = "Success"
  88     message = "not executed"
  89     error_message = ""
  90     prefix = ""
  91     suffix = ""
  92     row_class = QueryRow
  93
  94     def __init__(self, *args, **keywds):
  95         try:
  96             self.connection = keywds['connection']
  97         except KeyError:
  98             self.connection = connection
  99         try:
 100             self.diagnostics = keywds['diagnostics']
 101         except KeyError:
 102             self.diagnostics = 0
 103
 104         self.statement = self.prefix + self.__doc__ + self.suffix
 105         self.params = args
 106
 107     def __iter__(self):
 108         return IterationCursor(self, self.connection)
 109
 110     def __repr__(self):
 111         return "%s(message=%s)" % (self.__class__.__name__, self.message)
 112
 113     def cursor(self):
 114         return iter(self).cursor
 115
 116     def dump(self):
 117         for item in self:
 118             print item
 119
 120 class QueryGeneric(Query):
 121     def __init__(self, statement, *args, **keywds):
 122         Query.__init__(self, *args, **keywds)
 123         self.statement = statement,
 124
 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)
 135
 136     def next(self):
 137         return self.row_class(self.cursor)
 138
 139 class QuerySingle(Query, QueryRow):
 140     ignore_warnings = 0
 141     def __init__(self, *args, **keywds):
 142         message = self.MSG_FAILURE
 143         Query.__init__(self, *args, **keywds)
 144         try:
 145             self.single_cursor = Query.cursor(self)
 146         except MySQLdb.Warning:
 147             if not self.ignore_warnings:
 148                 raise
 149         self.row_class.__init__(self, self.cursor())
 150         object.__setattr__(self, "message", self.MSG_SUCCESS)
 151
 152     def cursor(self):
 153         return self.single_cursor
 154
 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()))
 159
 160     def process_row(self, row):
 161         return row
 162
 163 class QueryAllFirstItem(QueryAll):
 164     def process_row(self, row):
 165         return row[0]
 166
 167 class Create(QuerySingle):
 168     def __init__(self, *args, **keywds):
 169         try:
 170             QuerySingle.__init__(self, *args, **keywds)
 171         except StopIteration:
 172             self.message = self.MSG_SUCCESS
 173
 174 class Update(Create):
 175     pass
 176
 177 class Insert(Create):
 178     MSG_INTEGRITY_ERROR = "Couldn't insert: %s. "
 179
 180     def __init__(self, *args, **keywds):
 181         try:
 182             Create.__init__(self, *args, **keywds)
 183         except MySQLdb.IntegrityError, error_data:
 184             self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1]
 185             try:
 186                 self.total_count
 187             except AttributeError:
 188                 self.total_count = 0
 189
 190             raise MySQLdb.IntegrityError(self.error_message)
 191
 192         self.id = self.cursor().insert_id()
 193         try:
 194             self.total_count += self.cursor().rowcount
 195         except AttributeError:
 196             self.total_count = self.cursor().rowcount
 197
 198         if self.cursor().rowcount == 0:
 199             raise NoInsertionError
 200
 201 def _test(*args, **keywds):
 202     import doctest, sys
 203     doctest.testmod(sys.modules[__name__], *args, **keywds)
 204
 205 if __name__ == "__main__":
 206     if __debug__:
 207         _test()