我正在寻找正确的类型注释来归因于 celery 装饰器设置的 celery 任务@task。
假设我有一个函数run_task可以执行一些 celery 任务。
@task(name='adder')
def add(x, y):
return x + y
def run_task(taskfn: "?", *a, **kw):
# do something here
taskfn.apply_async(args=a, kwargs=kw)
...
>>> run_task(add, 1, 2) # usage
Run Code Online (Sandbox Code Playgroud)
我的困惑就在这里:
>>> from celery.task import Task
>>> from celery.local import PromiseProxy
>>> type(add).__name__
'PromiseProxy'
>>> isinstance(add, Task)
False
>>> isinstance(add, PromiseProxy)
True
Run Code Online (Sandbox Code Playgroud)
看起来像是PromiseProxy某种“未来”或“承诺”代理来保存任务(或类似的东西,我还没有深入研究它)。
这不会是一个问题,我可以轻松做到taskfn: PromiseProxy,但这样就taskfn.apply_async没有意义了,apply_async因为PromiseProxy. 所以,我的问题是,我应该按照要求去做type,还是按照鸭子打字的方式思考并继续Task?
我正在编写自定义 Django 迁移脚本。根据关于自定义迁移的django 文档,我应该能够使用我的模型 vis-a-vis apps.get_model()。但是,在尝试执行此操作时,出现以下错误:
AttributeError: type object 'MyModel' has no attribute 'objects'
Run Code Online (Sandbox Code Playgroud)
我认为这与应用程序注册表尚未准备好有关,但我不确定。
示例代码:
def do_thing(apps, schema_editor):
my_model = apps.get_model('app', 'MyModel')
objects_ = my_model.objects.filter(
some_field__isnull=True).prefetch_related(
'some_field__some_other_field') # exc raised here
class Migration(migrations.Migration):
atomic = False
dependencies = [
('app', '00xx_auto_xxx')
]
operations = [
migrations.RunPython(do_thing),
]
Run Code Online (Sandbox Code Playgroud)
apps.get_model()的返回值的简单打印语句显示如下:
<class '__fake__.MyModel'>。我不确定这是什么,是否是没有准备好的结果。
编辑:
我找不到任何资源来解释为什么我得到一个__fake__对象,所以我决定修改代码。我通过抢占appsargs使其工作,如下所示:
def do_thing(apps, schema_editor):
from django.apps import apps
my_model = apps.get_model('app', 'MyModel')
objects_ = my_model.objects.filter(
some_field__isnull=True).prefetch_related(
'some_field__some_other_field') …Run Code Online (Sandbox Code Playgroud) 我注意到一些带有/参数的函数签名。可以在以下位置找到一个示例collections.Counter.__init__():
def __init__(self, iterable=None, /, **kwds):
'''Create a new, empty Counter object. And if given, count elements
from an input iterable. Or, initialize the count from another mapping
of elements to their counts.
>>> c = Counter() # a new, empty counter
>>> c = Counter('gallahad') # a new counter from an iterable
>>> c = Counter({'a': 4, 'b': 2}) # a new counter from a mapping
>>> c = Counter(a=4, b=2) # a new counter …Run Code Online (Sandbox Code Playgroud) 您好,我需要比较 2 个字典,但在结果中,我需要知道结果来自哪个字典。
dict1 = {'name': 'Morgan', 'surename': 'Finch'}
dict2 = {'name': 'David', 'surename': 'Finch'}
Run Code Online (Sandbox Code Playgroud)
所以如果我与input_data.items() ^ response_data.items()结果比较将是这样的:
{('name','Morgan'),('name', 'David)}
Run Code Online (Sandbox Code Playgroud)
预期结果应该看起来像{'dict1': ('name','Morgan'), dict2: ('name', 'David')}
我不在乎什么数据结构只是我可以从它来的字典中知道。
super().__new__()在元类中重写时调用__new__包含以下函数签名:
class MyMeta(type):
def __new__(cls, name, bases, classdict):
clsobj = super().__new__(cls, name, bases, classdict)
Run Code Online (Sandbox Code Playgroud)
然而,当__new__在普通类中重写时,我们有以下内容:
class A:
def __new__(cls, *a, **kw):
clsobj = super().__new__(cls)
Run Code Online (Sandbox Code Playgroud)
将任何其他参数传递给super()inA.__new__将导致以下错误:
TypeError: object.__new__() takes exactly one argument (the type to instantiate)
Run Code Online (Sandbox Code Playgroud)
据我所知,在第二种情况下,我们正在处理object.__new__,而在第一种情况下,我们正在处理type.__new__。
我的问题是,为什么这些函数签名不同?为什么object.__new__只接受cls?
python ×5
celery ×1
collections ×1
django ×1
metaclass ×1
python-3.7 ×1
python-3.x ×1
super ×1
type-hinting ×1