TryFinally.py :  » Web-Frameworks » Aquarium » aquarium-2.3 » aquarium » util » Python Open Source

Home
Python Open Source
1.3.1.2 Python
2.Ajax
3.Aspect Oriented
4.Blog
5.Build
6.Business Application
7.Chart Report
8.Content Management Systems
9.Cryptographic
10.Database
11.Development
12.Editor
13.Email
14.ERP
15.Game 2D 3D
16.GIS
17.GUI
18.IDE
19.Installer
20.IRC
21.Issue Tracker
22.Language Interface
23.Log
24.Math
25.Media Sound Audio
26.Mobile
27.Network
28.Parser
29.PDF
30.Project Management
31.RSS
32.Search
33.Security
34.Template Engines
35.Test
36.UML
37.USB Serial
38.Web Frameworks
39.Web Server
40.Web Services
41.Web Unit
42.Wiki
43.Windows
44.XML
Python Open Source » Web Frameworks » Aquarium 
Aquarium » aquarium 2.3 » aquarium » util » TryFinally.py
"""This is a convenient way to deeply nest try/finally statements."""

__docformat__ = "restructuredtext"

# Created: Tue Oct 11 08:42:23 PDT 2005
# Author: Shannon -jj Behrens
# Email: jjinux@users.sourceforge.net
#
# Copyright (c) Shannon -jj Behrens.  All rights reserved.

from traceback import print_exc


def tryFinally(tasks, handleFinallyException=None):

    """This is a convenient way to deeply nest try/finally statements.

    It is appropriate for complicated resource initialization and destruction.
    For instance, if you have a list of 50 things that need to get intialized
    and later destructed via using try/finally (especially if you need to
    create the list dynamically) this function is appropriate.

    Given::

        tasks = [
            ((f_enter_0, enter_0_args, enter_0_kargs),
             (f_exit_0, exit_0_args, exit_0_kargs)),

            ((f_enter_1, enter_1_args, enter_1_kargs),
             (f_exit_1, exit_1_args, exit_1_kargs)),

            ((f_enter_2, enter_2_args, enter_2_kargs),
             (f_exit_2, exit_2_args, exit_2_kargs))
        ]

    Execute::

        f_enter_0(*enter_0_args, **enter_0_kargs)
        try:

            f_enter_1(*enter_1_args, **enter_1_kargs)
            try:

                f_enter_2(*enter_2_args, **enter_2_kargs)
                try:

                    pass

                finally:
                    try:
                        f_exit_2(*exit_2_args, **exit_2_kargs)
                    except Exception, e:
                        handleFinallyException(e)

            finally:
                try:
                    f_exit_1(*exit_1_args, **exit_1_kargs)
                except Exception, e:
                    handleFinallyException(e)

        finally:
            try:
                f_exit_0(*exit_0_args, **exit_0_kargs)
            except Exception, e:
                handleFinallyException(e)

    tasks
        See the example above.  Note that you can leave out parts of the tuples
        by passing shorter tuples.  For instance, here are two examples::

            # Second tuple missing.
            ((f_enter_2, enter_2_args, enter_2_kargs),)

            # Leave out args or args and kargs.
            ((f_enter_2,),
             (f_exit_2, exit_2_args))

        Don't forget that a tuple of 1 item is written ``(item,)``.  This is an
        amazingly easy thing to do.

    handleFinallyException(e)
        This is a callback that gets called if an exception, ``e``, is raised
        in a finally block.  By default, traceback.print_exc is called.

    """

    def defaultFinallyExceptionHandler(e):
        print_exc()

    if not len(tasks):
        return
    if not handleFinallyException:
        handleFinallyException = defaultFinallyExceptionHandler
    first, others = tasks[0], tasks[1:]
    ((f_enter, f_enter_args, f_enter_kargs),
     (f_exit, f_exit_args, f_exit_kargs)) = normalizeTask(first)

    f_enter(*f_enter_args, **f_enter_kargs)
    try:
        tryFinally(others, handleFinallyException)
    finally:
        try:
            f_exit(*f_exit_args, **f_exit_kargs)
        except Exception, e:
            handleFinallyException(e)


def getIndexOfTask(tasks, f_enter=None, f_exit=None):
    """Get the index of a task based on its enter or exit function.

    Once you have the index of the task, you can update the tasks list by
    putting another task before or after that index.  This is useful for
    subclasses updating a list created by a superclass.

    Raise IndexError if the index cannot be found.

    f_enter, f_exit
        You may specify the task using either is enter or exit function or
        both.

    """
    if not f_enter and not f_exit:
        raise ValueError("f_enter and f_exit can't both be None.")
    for (i, task) in enumerate(tasks):
        ((task_f_enter, f_enter_args, f_enter_kargs),
         (task_f_exit, f_exit_args, f_exit_kargs)) = normalizeTask(task)
        if ((not f_enter or f_enter == task_f_enter) and
            (not f_exit or f_exit == task_f_exit)):
            return i
    raise IndexError


def normalizeTask(task):

    """Given a task, return it after filling in all the blanks in the tuples.

    That is, return::

        ((f_enter, f_enter_args, f_enter_kargs),
         (f_exit, f_exit_args, f_exit_kargs))

    """

    def castTwoParts(first):
        lenFirst = len(first)
        default = ((), ())
        max = len(default)
        if lenFirst > max:
            raise ValueError("""\
A task is a tuple of the form (enterTuple, exitTuple).""", first)
        return first + default[lenFirst:]

    def doNothing(*args, **kargs):
        pass

    def castFunctionArgsKargs(fTuple):
        lenFTuple = len(fTuple)
        default = (doNothing, (), {})
        max = len(default)
        if lenFTuple > max:
            raise ValueError("""\
Each tuple in a tasks is a pair of tuples that look like (f, args, kargs).""",
                             fTuple)
        return fTuple + default[lenFTuple:]

    task = castTwoParts(task)
    return (castFunctionArgsKargs(task[0]),
            castFunctionArgsKargs(task[1]))


if __name__ == '__main__':

    from cStringIO import StringIO

    def printEverything(*args, **kargs): print >>buf, `args`, `kargs`
    def refuseArgs(): print >>buf, "refused args"
    def raiseValueError(): raise ValueError
    def finallyExceptionHandler(e): print >>buf, "caught exception in finally"

    tasks = [
        ((printEverything, ("enter_0_args",), {"in": "in"}),
         (printEverything, ("exit_0_args",))),

        ((printEverything,),
         (raiseValueError,)),

        ((refuseArgs,),)
    ]

    result = """\
('enter_0_args',) {'in': 'in'}
() {}
refused args
caught exception in finally
('exit_0_args',) {}
"""

    buf = StringIO()
    tryFinally(tasks, finallyExceptionHandler)
    assert buf.getvalue() == result
    assert getIndexOfTask(tasks, f_enter=printEverything) == 0
    assert getIndexOfTask(tasks, f_enter=printEverything,
                          f_exit=raiseValueError) == 1
    assert getIndexOfTask(tasks, f_exit=raiseValueError) == 1
www.java2java.com | Contact Us
Copyright 2009 - 12 Demo Source and Support. All rights reserved.
All other trademarks are property of their respective owners.