Logo Search packages:      
Sourcecode: sqlalchemy version File versions  Download package

def sqlalchemy::engine::base::ResultProxy::__init__ (   self,
  executioncontext = None,
  typemap = None 

ResultProxy objects are constructed via the execute() method on SQLEngine.

Definition at line 545 of file base.py.

00545                                                                                        :
        """ResultProxy objects are constructed via the execute() method on SQLEngine."""
        self.connection = connection
        self.dialect = engine.dialect
        self.cursor = cursor
        self.engine = engine
        self.closed = False
        if executioncontext is not None:
            self.__executioncontext = executioncontext
            self.rowcount = executioncontext.get_rowcount(cursor)
            self.rowcount = cursor.rowcount
        self.__key_cache = {}
        self.__echo = engine.echo == 'debug'
        metadata = cursor.description
        self.props = {}
        self.keys = []
        i = 0
        if metadata is not None:
            for item in metadata:
                # sqlite possibly prepending table name to colnames so strip
                colname = item[0].split('.')[-1].lower()
                if typemap is not None:
                    rec = (typemap.get(colname, types.NULLTYPE), i)
                    rec = (types.NULLTYPE, i)
                if rec[0] is None:
                    raise DBAPIError("None for metadata " + colname)
                if self.props.setdefault(colname, rec) is not rec:
                    self.props[colname] = (ResultProxy.AmbiguousColumn(colname), 0)
                self.props[i] = rec
    def _executioncontext(self):
            return self.__executioncontext
        except AttributeError:
            raise exceptions.InvalidRequestError("This ResultProxy does not have an execution context with which to complete this operation.  Execution contexts are not generated for literal SQL execution.")
    executioncontext = property(_executioncontext)
    def close(self):
        """close this ResultProxy, and the underlying DBAPI cursor corresponding to the execution.
        If this ResultProxy was generated from an implicit execution, the underlying Connection will
        also be closed (returns the underlying DBAPI connection to the connection pool.)
        This method is also called automatically when all result rows are exhausted."""
        if not self.closed:
            self.closed = True
            if self.connection.should_close_with_result and self.dialect.supports_autoclose_results:
    def _convert_key(self, key):
        """given a key, which could be a ColumnElement, string, etc., matches it to the 
        appropriate key we got from the result set's metadata; then cache it locally for quick re-access."""
            return self.__key_cache[key]
        except KeyError:
            if isinstance(key, sql.ColumnElement):
                    rec = self.props[key._label.lower()]
                except KeyError:
                        rec = self.props[key.key.lower()]
                    except KeyError:
#                        rec = self.props[key.name.lower()]
                            rec = self.props[key.name.lower()]
                        except KeyError:
                            raise exceptions.NoSuchColumnError("Could not locate column in row for column '%s'" % str(key))
            elif isinstance(key, str):
                rec = self.props[key.lower()]
                rec = self.props[key]
            self.__key_cache[key] = rec
            return rec
    def _has_key(self, row, key):
            return True
        except exceptions.NoSuchColumnError:
            return False
    def _get_col(self, row, key):
        rec = self._convert_key(key)
        return rec[0].dialect_impl(self.dialect).convert_result_value(row[rec[1]], self.dialect)
    def __iter__(self):
        while True:
            row = self.fetchone()
            if row is None:
                raise StopIteration
                yield row
    def last_inserted_ids(self):
        """return last_inserted_ids() from the underlying ExecutionContext.
        See ExecutionContext for details."""
        return self.executioncontext.last_inserted_ids()
    def last_updated_params(self):
        """return last_updated_params() from the underlying ExecutionContext.
        See ExecutionContext for details."""
        return self.executioncontext.last_updated_params()
    def last_inserted_params(self):
        """return last_inserted_params() from the underlying ExecutionContext.
        See ExecutionContext for details."""
        return self.executioncontext.last_inserted_params()
    def lastrow_has_defaults(self):
        """return lastrow_has_defaults() from the underlying ExecutionContext.
        See ExecutionContext for details."""
        return self.executioncontext.lastrow_has_defaults()
    def supports_sane_rowcount(self):
        """return supports_sane_rowcount() from the underlying ExecutionContext.
        See ExecutionContext for details."""
        return self.executioncontext.supports_sane_rowcount()
    def fetchall(self):
        """fetch all rows, just like DBAPI cursor.fetchall()."""
        l = []
        for row in self.cursor.fetchall():
            l.append(RowProxy(self, row))
        return l
    def fetchone(self):
        """fetch one row, just like DBAPI cursor.fetchone()."""
        row = self.cursor.fetchone()
        if row is not None:
            return RowProxy(self, row)
            # controversy!  can we auto-close the cursor after results are consumed ?
            # what if the returned rows are still hanging around, and are "live" objects 
            # and not just plain tuples ?
            return None

    def scalar(self):
        """fetch the first column of the first row, and close the result set."""
        row = self.cursor.fetchone()
            if row is not None:
                return RowProxy(self, row)[0]
                return None
class RowProxy(object):

Generated by  Doxygen 1.6.0   Back to index