如何在Python中将变量放入堆栈/上下文中

发布于 2024-07-24 22:10:14 字数 1545 浏览 3 评论 0 原文

本质上,我想将一个变量放在堆栈上,堆栈上该部分下方的所有调用都可以访问该变量,直到块退出。 在Java中,我将使用带有支持方法的本地静态线程来解决这个问题,然后可以从方法中访问该方法。

典型示例:您收到一个请求,并打开一个数据库连接。 在请求完成之前,您希望所有代码都使用此数据库连接。 完成并关闭请求后,您可以关闭数据库连接。

我需要这个的目的是一个报告生成器。 每个报告由多个部分组成,每个部分可以依赖于不同的计算,有时不同的部分部分依赖于相同的计算。 由于我不想重复繁重的计算,因此我需要缓存它们。 我的想法是用缓存装饰器来装饰方法。 缓存根据方法名称和模块及其参数创建一个 id,查看堆栈变量中是否已计算出该 id,如果没有,则执行该方法。

我将尝试通过展示我当前的实现来澄清这一点。 我想做的就是简化那些实现计算的代码。

首先,我有中央缓存访问对象,我将其称为 MathContext:

class MathContext(object):
    def __init__(self, fn): 
        self.fn = fn
        self.cache = dict()
    def get(self, calc_config):
        id = create_id(calc_config)
        if id not in self.cache:
            self.cache[id] = calc_config.exec(self)
        return self.cache[id]

fn 参数是创建上下文所关联的文件名,可以从中读取数据进行计算。

然后我们有计算类:

 class CalcBase(object):
     def exec(self, math_context):
         raise NotImplementedError

这是一个愚蠢的斐波那契示例。 这些方法实际上不是递归的,它们适用于大量数据,但它可以演示如何依赖其他计算:

class Fibonacci(CalcBase):
    def __init__(self, n): self.n = n
    def exec(self, math_context):
        if self.n < 2: return 1
        a = math_context.get(Fibonacci(self.n-1))
        b = math_context.get(Fibonacci(self.n-2))
        return a+b

我希望斐波那契只是一个装饰方法:

@cache
def fib(n):
    if n<2: return 1
    return fib(n-1)+fib(n-2)

以 math_context 为例,当math_context 超出了范围,所有缓存的值也超出了范围。 我想要装饰器也有同样的东西。 IE。 在 X 点,@cache 缓存的所有内容都被取消引用以进行 gced。

In essence, I want to put a variable on the stack, that will be reachable by all calls below that part on the stack until the block exits. In Java I would solve this using a static thread local with support methods, that then could be accessed from methods.

Typical example: you get a request, and open a database connection. Until the request is complete, you want all code to use this database connection. After finishing and closing the request, you close the database connection.

What I need this for, is a report generator. Each report consist of multiple parts, each part can rely on different calculations, sometimes different parts relies in part on the same calculation. As I don't want to repeat heavy calculations, I need to cache them. My idea is to decorate methods with a cache decorator. The cache creates an id based on the method name and module, and it's arguments, looks if it has this allready calculated in a stack variable, and executes the method if not.

I will try and clearify by showing my current implementation. Want I want to do is to simplify the code for those implementing calculations.

First, I have the central cache access object, which I call MathContext:

class MathContext(object):
    def __init__(self, fn): 
        self.fn = fn
        self.cache = dict()
    def get(self, calc_config):
        id = create_id(calc_config)
        if id not in self.cache:
            self.cache[id] = calc_config.exec(self)
        return self.cache[id]

The fn argument is the filename the context is created in relation to, from where data can be read to be calculated.

Then we have the Calculation class:

 class CalcBase(object):
     def exec(self, math_context):
         raise NotImplementedError

And here is a stupid Fibonacci example. Non of the methods are actually recursive, they work on large sets of data instead, but it works to demonstrate how you would depend on other calculations:

class Fibonacci(CalcBase):
    def __init__(self, n): self.n = n
    def exec(self, math_context):
        if self.n < 2: return 1
        a = math_context.get(Fibonacci(self.n-1))
        b = math_context.get(Fibonacci(self.n-2))
        return a+b

What I want Fibonacci to be instead, is just a decorated method:

@cache
def fib(n):
    if n<2: return 1
    return fib(n-1)+fib(n-2)

With the math_context example, when math_context goes out of scope, so does all it's cached values. I want the same thing for the decorator. Ie. at point X, everything cached by @cache is dereferrenced to be gced.

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(5

提笔落墨 2024-07-31 22:10:14

我继续做了一些可以满足你想要的东西。 它既可以用作装饰器,也可以用作上下文管理器:

from __future__ import with_statement
try:
    import cPickle as pickle
except ImportError:
    import pickle


class cached(object):
    """Decorator/context manager for caching function call results.
    All results are cached in one dictionary that is shared by all cached
    functions.

    To use this as a decorator:
        @cached
        def function(...):
            ...

    The results returned by a decorated function are not cleared from the
    cache until decorated_function.clear_my_cache() or cached.clear_cache()
    is called

    To use this as a context manager:

        with cached(function) as function:
            ...
            function(...)
            ...

    The function's return values will be cleared from the cache when the
    with block ends

    To clear all cached results, call the cached.clear_cache() class method
    """

    _CACHE = {}

    def __init__(self, fn):
        self._fn = fn

    def __call__(self, *args, **kwds):
        key = self._cache_key(*args, **kwds)
        function_cache = self._CACHE.setdefault(self._fn, {})
        try:
            return function_cache[key]
        except KeyError:
            function_cache[key] = result = self._fn(*args, **kwds)
            return result

    def clear_my_cache(self):
        """Clear the cache for a decorated function
        """
        try:
            del self._CACHE[self._fn]
        except KeyError:
            pass # no cached results

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        self.clear_my_cache()

    def _cache_key(self, *args, **kwds):
        """Create a cache key for the given positional and keyword
        arguments. pickle.dumps() is used because there could be
        unhashable objects in the arguments, but passing them to 
        pickle.dumps() will result in a string, which is always hashable.

        I used this to make the cached class as generic as possible. Depending
        on your requirements, other key generating techniques may be more
        efficient
        """
        return pickle.dumps((args, sorted(kwds.items())), pickle.HIGHEST_PROTOCOL)

    @classmethod
    def clear_cache(cls):
        """Clear everything from all functions from the cache
        """
        cls._CACHE = {}


if __name__ == '__main__':
    # used as decorator
    @cached
    def fibonacci(n):
        print "calculating fibonacci(%d)" % n
        if n == 0:
            return 0
        if n == 1:
            return 1
        return fibonacci(n - 1) + fibonacci(n - 2)

    for n in xrange(10):
        print 'fibonacci(%d) = %d' % (n, fibonacci(n))


    def lucas(n):
        print "calculating lucas(%d)" % n
        if n == 0:
            return 2
        if n == 1:
            return 1
        return lucas(n - 1) + lucas(n - 2)

    # used as context manager
    with cached(lucas) as lucas:
        for i in xrange(10):
            print 'lucas(%d) = %d' % (i, lucas(i))

    for n in xrange(9, -1, -1):
        print 'fibonacci(%d) = %d' % (n, fibonacci(n))

    cached.clear_cache()

    for n in xrange(9, -1, -1):
        print 'fibonacci(%d) = %d' % (n, fibonacci(n))

I went ahead and made something that might just do what you want. It can be used as both a decorator and a context manager:

from __future__ import with_statement
try:
    import cPickle as pickle
except ImportError:
    import pickle


class cached(object):
    """Decorator/context manager for caching function call results.
    All results are cached in one dictionary that is shared by all cached
    functions.

    To use this as a decorator:
        @cached
        def function(...):
            ...

    The results returned by a decorated function are not cleared from the
    cache until decorated_function.clear_my_cache() or cached.clear_cache()
    is called

    To use this as a context manager:

        with cached(function) as function:
            ...
            function(...)
            ...

    The function's return values will be cleared from the cache when the
    with block ends

    To clear all cached results, call the cached.clear_cache() class method
    """

    _CACHE = {}

    def __init__(self, fn):
        self._fn = fn

    def __call__(self, *args, **kwds):
        key = self._cache_key(*args, **kwds)
        function_cache = self._CACHE.setdefault(self._fn, {})
        try:
            return function_cache[key]
        except KeyError:
            function_cache[key] = result = self._fn(*args, **kwds)
            return result

    def clear_my_cache(self):
        """Clear the cache for a decorated function
        """
        try:
            del self._CACHE[self._fn]
        except KeyError:
            pass # no cached results

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        self.clear_my_cache()

    def _cache_key(self, *args, **kwds):
        """Create a cache key for the given positional and keyword
        arguments. pickle.dumps() is used because there could be
        unhashable objects in the arguments, but passing them to 
        pickle.dumps() will result in a string, which is always hashable.

        I used this to make the cached class as generic as possible. Depending
        on your requirements, other key generating techniques may be more
        efficient
        """
        return pickle.dumps((args, sorted(kwds.items())), pickle.HIGHEST_PROTOCOL)

    @classmethod
    def clear_cache(cls):
        """Clear everything from all functions from the cache
        """
        cls._CACHE = {}


if __name__ == '__main__':
    # used as decorator
    @cached
    def fibonacci(n):
        print "calculating fibonacci(%d)" % n
        if n == 0:
            return 0
        if n == 1:
            return 1
        return fibonacci(n - 1) + fibonacci(n - 2)

    for n in xrange(10):
        print 'fibonacci(%d) = %d' % (n, fibonacci(n))


    def lucas(n):
        print "calculating lucas(%d)" % n
        if n == 0:
            return 2
        if n == 1:
            return 1
        return lucas(n - 1) + lucas(n - 2)

    # used as context manager
    with cached(lucas) as lucas:
        for i in xrange(10):
            print 'lucas(%d) = %d' % (i, lucas(i))

    for n in xrange(9, -1, -1):
        print 'fibonacci(%d) = %d' % (n, fibonacci(n))

    cached.clear_cache()

    for n in xrange(9, -1, -1):
        print 'fibonacci(%d) = %d' % (n, fibonacci(n))
赠意 2024-07-31 22:10:14

这个问题似乎是两个问题

  • a)共享数据库连接
  • b)缓存/记忆

b)你已经回答了自己

a)我似乎不明白为什么你需要将它放在堆栈上?
你可以做其中之一

  1. 你可以使用类和连接
    可能是它的属性,
  2. 你可以装饰你的所有功能
    这样他们就可以从
    中心位置
  3. 每个函数都可以显式使用一个
    全局连接方法,
  4. 您可以创建连接并传递
    围绕它,或创建一个上下文
    对象并传递
    上下文,连接可以是一部分
    上下文

this question seems to be two question

  • a) sharing db connection
  • b) caching/Memoizing

b) you have answered yourselves

a) I don't seem to understand why you need to put it on stack?
you can do one of these

  1. you can use a class and connection
    could be attribute of it
  2. you can decorate all your function
    so that they get a connection from
    central location
  3. each function can explicitly use a
    global connection method
  4. you can create a connection and pass
    around it, or create a context
    object and pass around
    context,connection can be a part of
    context

etc, etc

柒七 2024-07-31 22:10:14

您可以使用封装在 getter 函数中的全局变量:

def getConnection():
    global connection
    if connection:
        return connection
    connection=createConnection()
    return connection

You could use a global variable wrapped in a getter function:

def getConnection():
    global connection
    if connection:
        return connection
    connection=createConnection()
    return connection
梦魇绽荼蘼 2024-07-31 22:10:14

“你收到一个请求,并打开一个数据库连接......你关闭数据库连接。”

这就是对象的用途。 创建连接对象,将其传递给其他对象,然后在完成后关闭它。 全局变量不合适。 只需将值作为参数传递给正在执行该工作的其他对象即可。

“每个报告由多个部分组成,每个部分可以依赖于不同的计算,有时不同的部分部分依赖于相同的计算......我需要缓存它们”

这就是对象的用途。 创建包含有用计算结果的字典,并将其在报表部件之间传递。

你不需要搞乱“堆栈变量”、“静态线程本地”或类似的东西。
只需将普通变量参数传递给普通方法函数即可。 你会快乐很多。


class MemoizedCalculation( object ):
    pass

class Fibonacci( MemoizedCalculation ):
    def __init__( self ):
       self.cache= { 0: 1, 1: 1 }
    def __call__( self, arg ):
       if arg not in self.cache:
           self.cache[arg]= self(arg-1) + self(arg-2)
       return self.cache[arg]

class MathContext( object ):
    def __init__( self ):
        self.fibonacci = Fibonacci()

您可以像这样使用它

>>> mc= MathContext()
>>> mc.fibonacci( 4 )
5

您可以定义任意数量的计算并将它们全部折叠到单个容器对象中。

如果需要,您可以将 MathContext 变成正式的上下文管理器,以便它与 with 语句一起使用。 将这两个方法添加到 MathContext。

def __enter__( self ):
    print "Initialize"
    return self
def __exit__( self, type_, value, traceback ):
    print "Release"

然后你就可以这样做。

with  MathContext() as mc:
    print mc.fibonacci( 4 )

with 语句的末尾,您可以保证调用 __exit__ 方法。

"you get a request, and open a database connection.... you close the database connection."

This is what objects are for. Create the connection object, pass it to other objects, and then close it when you're done. Globals are not appropriate. Simply pass the value around as a parameter to the other objects that are doing the work.

"Each report consist of multiple parts, each part can rely on different calculations, sometimes different parts relies in part on the same calculation.... I need to cache them"

This is what objects are for. Create a dictionary with useful calculation results and pass that around from report part to report part.

You don't need to mess with "stack variables", "static thread local" or anything like that.
Just pass ordinary variable arguments to ordinary method functions. You'll be a lot happier.


class MemoizedCalculation( object ):
    pass

class Fibonacci( MemoizedCalculation ):
    def __init__( self ):
       self.cache= { 0: 1, 1: 1 }
    def __call__( self, arg ):
       if arg not in self.cache:
           self.cache[arg]= self(arg-1) + self(arg-2)
       return self.cache[arg]

class MathContext( object ):
    def __init__( self ):
        self.fibonacci = Fibonacci()

You can use it like this

>>> mc= MathContext()
>>> mc.fibonacci( 4 )
5

You can define any number of calculations and fold them all into a single container object.

If you want, you can make the MathContext into a formal Context Manager so that it work with the with statement. Add these two methods to MathContext.

def __enter__( self ):
    print "Initialize"
    return self
def __exit__( self, type_, value, traceback ):
    print "Release"

Then you can do this.

with  MathContext() as mc:
    print mc.fibonacci( 4 )

At the end of the with statement, you can guaranteed that the __exit__ method was called.

嗫嚅 2024-07-31 22:10:14

同时(python 3.7及更高版本)同样的问题找到了另一个正确的解决方案: from contextvars import ContextVar

这里是 PEP

https://peps.python.org/pep-0567/

meanwhile (python 3.7 and later) the same problem has found another proper solution: from contextvars import ContextVar

here is the PEP

https://peps.python.org/pep-0567/

~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文