Back to index

python-biopython  1.60
DocSQL.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 #
00003 # Copyright 2002-2003 by Michael Hoffman.  All rights reserved.
00004 # This code is part of the Biopython distribution and governed by its
00005 # license.  Please see the LICENSE file that should have been included
00006 # as part of this package.
00007 
00008 """
00009 Bio.DocSQL: easy access to DB API databases.
00010 
00011 >>> import os
00012 >>> import MySQLdb
00013 >>> from Bio import DocSQL
00014 >>> db=MySQLdb.connect(passwd='', db='test')
00015 >>> class CreatePeople(DocSQL.Create):
00016 ...     '''
00017 ...     CREATE TEMPORARY TABLE people
00018 ...     (id INT UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT,
00019 ...     last_name TINYTEXT,
00020 ...     first_name TINYTEXT)
00021 ...     '''
00022 ...
00023 >>> CreatePeople(connection=db)
00024 CreatePeople(message=Success)
00025 """
00026 
00027 __version__ = "$Revision: 1.13 $"
00028 # $Source: /home/bartek/cvs2bzr/biopython_fastimport/cvs_repo/biopython/Bio/DocSQL.py,v $
00029 
00030 import sys
00031 
00032 from Bio import MissingPythonDependencyError
00033 
00034 try:
00035     import MySQLdb
00036 except:
00037     raise MissingPythonDependencyError("Install MySQLdb if you want to use "
00038                                        "Bio.DocSQL.")
00039 
00040 connection = None
00041 
00042 class NoInsertionError(Exception):
00043     pass
00044 
00045 def _check_is_public(name):
00046     if name[:6] == "_names":
00047         raise AttributeError
00048     
00049 class QueryRow(list):
00050     def __init__(self, cursor):
00051         try:
00052             row = cursor.fetchone()
00053             super(QueryRow, self).__init__(row)
00054         except TypeError:
00055             raise StopIteration
00056 
00057         object.__setattr__(self, "_names", [x[0] for x in cursor.description]) # FIXME: legacy
00058         object.__setattr__(self, "_names_hash", {})
00059         
00060         for i, name in enumerate(self._names):
00061             self._names_hash[name] = i
00062 
00063     def __getattr__(self, name):
00064         _check_is_public(name)
00065         try:
00066             return self[self._names_hash[name]]
00067         except (KeyError, AttributeError):
00068             raise AttributeError("'%s' object has no attribute '%s'" \
00069                                  % (self.__class__.__name__, name))
00070 
00071     def __setattr__(self, name, value):
00072         try:
00073             self._names_hash
00074         except AttributeError:
00075             return object.__setattr__(self, name, value)
00076             
00077         _check_is_public(name)
00078         try:
00079             index = self._names_hash[name]
00080             self[index] = value
00081         except KeyError:
00082             return object.__setattr__(self, name, value)
00083 
00084 class Query(object):
00085     """
00086     SHOW TABLES
00087     """
00088     MSG_FAILURE = "Failure"
00089     MSG_SUCCESS = "Success"
00090     message = "not executed"
00091     error_message = ""
00092     prefix = ""
00093     suffix = ""
00094     row_class = QueryRow
00095 
00096     def __init__(self, *args, **keywds):
00097         try:
00098             self.connection = keywds['connection']
00099         except KeyError:
00100             self.connection = connection
00101         try:
00102             self.diagnostics = keywds['diagnostics']
00103         except KeyError:
00104             self.diagnostics = 0
00105 
00106         self.statement = self.prefix + self.__doc__ + self.suffix
00107         self.params = args
00108 
00109     def __iter__(self):
00110         return IterationCursor(self, self.connection)
00111 
00112     def __repr__(self):
00113         return "%s(message=%s)" % (self.__class__.__name__, self.message)
00114 
00115     def cursor(self):
00116         return iter(self).cursor
00117 
00118     def dump(self):
00119         for item in self:
00120             print item
00121 
00122 class QueryGeneric(Query):
00123     def __init__(self, statement, *args, **keywds):
00124         Query.__init__(self, *args, **keywds)
00125         self.statement = statement,
00126 
00127 class IterationCursor(object):
00128     def __init__(self, query, connection=connection):
00129         if connection is None:
00130             raise TypeError("database connection is None")
00131         self.cursor = connection.cursor()
00132         self.row_class = query.row_class
00133         if query.diagnostics:
00134             print >>sys.stderr, query.statement
00135             print >>sys.stderr, query.params
00136         self.cursor.execute(query.statement, query.params)
00137 
00138     def next(self):
00139         return self.row_class(self.cursor)
00140 
00141 class QuerySingle(Query, QueryRow):
00142     ignore_warnings = 0
00143     def __init__(self, *args, **keywds):
00144         message = self.MSG_FAILURE
00145         Query.__init__(self, *args, **keywds)
00146         try:
00147             self.single_cursor = Query.cursor(self)
00148         except MySQLdb.Warning:
00149             if not self.ignore_warnings:
00150                 raise
00151         self.row_class.__init__(self, self.cursor())
00152         object.__setattr__(self, "message", self.MSG_SUCCESS)
00153 
00154     def cursor(self):
00155         return self.single_cursor
00156 
00157 class QueryAll(list, Query):
00158     def __init__(self, *args, **keywds):
00159         Query.__init__(self, *args, **keywds)
00160         list.__init__(self, map(self.process_row, self.cursor().fetchall()))
00161 
00162     def process_row(self, row):
00163         return row
00164 
00165 class QueryAllFirstItem(QueryAll):
00166     def process_row(self, row):
00167         return row[0]
00168 
00169 class Create(QuerySingle):
00170     def __init__(self, *args, **keywds):
00171         try:
00172             QuerySingle.__init__(self, *args, **keywds)
00173         except StopIteration:
00174             self.message = self.MSG_SUCCESS
00175 
00176 class Update(Create):
00177     pass
00178 
00179 class Insert(Create):
00180     MSG_INTEGRITY_ERROR = "Couldn't insert: %s. "
00181     
00182     def __init__(self, *args, **keywds):
00183         try:
00184             Create.__init__(self, *args, **keywds)
00185         except MySQLdb.IntegrityError, error_data:
00186             self.error_message += self.MSG_INTEGRITY_ERROR % error_data[1]
00187             try:
00188                 self.total_count
00189             except AttributeError:
00190                 self.total_count = 0
00191             
00192             raise MySQLdb.IntegrityError(self.error_message)
00193             
00194         self.id = self.cursor().insert_id()
00195         try:
00196             self.total_count += self.cursor().rowcount
00197         except AttributeError:
00198             self.total_count = self.cursor().rowcount
00199 
00200         if self.cursor().rowcount == 0:
00201             raise NoInsertionError
00202 
00203 def _test(*args, **keywds):
00204     import doctest, sys
00205     doctest.testmod(sys.modules[__name__], *args, **keywds)
00206 
00207 if __name__ == "__main__":
00208     if __debug__:
00209         _test()