python之Django/Python 中的 Celery 任务最佳实践

考虑一个需要异步完成一些繁重工作的函数。调用客户端可以立即收到缓存版本，或者收到数字正在处理的响应(对客户端的有效响应)。

以下代码片段是否是此模式的合理实现？

from django.core import cache 
from proj.celery import app 
 
class SomeModel(models.Model): 
    # ... 
    def get_crunched_numbers(self): 
        cache_key = 'foo:{}'.format(self.id) 
        res = cache.get(cache_key) 
        if not res: 
            @app.task 
            def gen_crunched_numbers(): 
                res = do_heavy_lifting() 
                cache.set(cache_key, res) 
                return res 
            gen_crunched_numbers.delay() 
            return 'crunching... come back later' 
        else: 
            return res 

是否有更好的替代方案来运行像这样的 Celery 任务，同时将所有逻辑包含在一段代码中？

编辑:如评论中所述，此代码甚至不起作用。因此，我们非常感谢任何关于漂亮图案的建议。

请您参考如下方法:

您的代码看起来很困惑。为什么不在类之外定义 celery 任务函数并像这样调用它:

from django.core import cache 
from proj.celery import app 
 
class SomeModel(models.Model): 
    # ... 
    def get_crunched_numbers(self): 
        cache_key = 'foo:{}'.format(self.id) 
        res = cache.get(cache_key) 
        if not res: 
            gen_crunched_numbers.delay(cache_key) 
            return 'crunching... come back later' 
        else: 
            return res 
 
@app.task 
def gen_crunched_numbers(cache_key): 
    res = do_heavy_lifting() 
    cache.set(cache_key, res) 
    return res 

此外，我通常更喜欢使用 bind=True 创建任务:

@app.task(bind=True) 
def gen_crunched_numbers(self, cache_key): 
    res = do_heavy_lifting() 
    cache.set(cache_key, res) 
    return res 

这让我可以访问 task context通过 self.request。例如，根据函数是通过 celery 调用还是直接调用来更改行为:

@app.task(bind=True) 
def gen_crunched_numbers(self, cache_key): 
    res = do_heavy_lifting() 
    cache.set(cache_key, res) 
    if self.request.called_directly: 
        return res 
    else: 
        return { 'result': res, 'cache': cache_key }