celery 运行时 （from . import async, base）SyntaxError: invalid syntax 异常的解决方案

今天，有一个任务需要异步执行。发现 celery 框架不错，于是尝试了一下。
按照官方文档中的示例，执行的时候出现了一个异常：

$ celery -A tasks worker --loglevel=info

[2018-11-16 18:16:52,668: CRITICAL/MainProcess] Unrecoverable error: SyntaxError(''invalid syntax'', (''/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/backends/redis.py'', 22, 19, ''from . import async, base\n''))
Traceback (most recent call last):
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/kombu/utils/objects.py", line 42, in __get__
    return obj.__dict__[self.__name__]
KeyError: ''backend''

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/worker/worker.py", line 205, in start
    self.blueprint.start(self)
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/bootsteps.py", line 115, in start
    self.on_start()
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/apps/worker.py", line 139, in on_start
    self.emit_banner()
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/apps/worker.py", line 154, in emit_banner
    '' \n'', self.startup_info(artlines=not use_image))),
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/apps/worker.py", line 217, in startup_info
    results=self.app.backend.as_uri(),
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/kombu/utils/objects.py", line 44, in __get__
    value = obj.__dict__[self.__name__] = self.__get(obj)
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/app/base.py", line 1196, in backend
    return self._get_backend()
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/app/base.py", line 914, in _get_backend
    self.loader)
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/app/backends.py", line 70, in by_url
    return by_name(backend, loader), url
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/app/backends.py", line 50, in by_name
    cls = symbol_by_name(backend, aliases)
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/kombu/utils/imports.py", line 56, in symbol_by_name
    module = imp(module_name, package=package, **kwargs)
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1006, in _gcd_import
  File "<frozen importlib._bootstrap>", line 983, in _find_and_load
  File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 724, in exec_module
  File "<frozen importlib._bootstrap_external>", line 860, in get_code
  File "<frozen importlib._bootstrap_external>", line 791, in source_to_code
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/celery/backends/redis.py", line 22
    from . import async, base
                      ^
SyntaxError: invalid syntax

根据错误提示，发现是文件celery/backends/redis.py的22行出现问题：from . import async, base。后来，网上查了一下，发现是 celery 4.2.1 版本有问题。

解决方案

从 master 分支重新获取，执行命令如下：
pip install --upgrade https://github.com/celery/celery/tarball/master

命令执行后，发现版本重置到了 celery-4.2.0。
执行过程：

$ pip install --upgrade https://github.com/celery/celery/tarball/master


Looking in indexes: http://mirrors.aliyun.com/pypi/simple/
Collecting https://github.com/celery/celery/tarball/master
  Downloading https://github.com/celery/celery/tarball/master
     / 1.9MB 856kB/s
Requirement already satisfied, skipping upgrade: pytz>dev in /Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages (from celery==4.2.0) (2018.7)
Requirement already satisfied, skipping upgrade: billiard<3.6.0,>=3.5.0.2 in /Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages (from celery==4.2.0) (3.5.0.4)
Requirement already satisfied, skipping upgrade: kombu<5.0,>=4.2.0 in /Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages (from celery==4.2.0) (4.2.1)
Requirement already satisfied, skipping upgrade: amqp<3.0,>=2.1.4 in /Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages (from kombu<5.0,>=4.2.0->celery==4.2.0) (2.3.2)
Requirement already satisfied, skipping upgrade: vine>=1.1.3 in /Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages (from amqp<3.0,>=2.1.4->kombu<5.0,>=4.2.0->celery==4.2.0) (1.1.4)
Building wheels for collected packages: celery
  Running setup.py bdist_wheel for celery ... done
  Stored in directory: /private/var/folders/nn/nm7dqchn0rzcs0f2l2mpjkq40000gn/T/pip-ephem-wheel-cache-lpfank45/wheels/77/5f/be/3607e48004e6496d18a500339472d42f85c7a2446157726958
Successfully built celery
Installing collected packages: celery
  Found existing installation: celery 4.2.1
    Uninstalling celery-4.2.1:
      Successfully uninstalled celery-4.2.1
Successfully installed celery-4.2.0

调用时 redis 异常

另外，在使用 celery + redis 调用时，发生了如下异常：

File "/Users/Allert/.local/share/virtualenvs/testpure-2oTTuBuP/lib/python3.7/site-packages/redis/_compat.py", line 94, in iteritems
    return x.iteritems()
AttributeError: ''float'' object has no attribute ''iteritems''

从错误提示中，可以看到是 redis 中的文件发生异常。
这貌似是 redis 3.0.1 的一个兼容问题。

解决方案：重新安装 2.10.6版本。
命令如下：pip install redis==2.10.6

参考：

https://stackoverflow.com/questions/53322425/celery-critical-mainprocess-unrecoverable-error-attributeerrorfloat-object

https://www.pyget.cn/p/186098

protected void checkCassandraException(Exception e) {
    _mexceptions.incr();
    if (e instanceof AlreadyExistsException ||
            e instanceof AuthenticationException ||
            e instanceof DriverException ||
            e instanceof DriverInternalError ||
            e instanceof InvalidConfigurationInQueryException ||
            e instanceof InvalidQueryException ||
            e instanceof InvalidTypeException ||
            e instanceof QueryExecutionException ||
            e instanceof QueryTimeoutException ||
            e instanceof QueryValidationException ||
            e instanceof ReadTimeoutException ||
            e instanceof SyntaxError ||
            e instanceof TraceRetrievalException ||
            e instanceof TruncateException ||
            e instanceof UnauthorizedException ||
            e instanceof UnavailableException ||
            e instanceof ReadTimeoutException ||
            e instanceof WriteTimeoutException) {
        throw new ReportedFailedException(e);
    } else {
        throw new RuntimeException(e);
    }
}

public void execute(InputStream commandStream) throws IOException {
    byte[] commandBuffer = new byte[commandStream.available()];
    IoUtils.readFully(commandStream,commandBuffer);

    Cluster cluster = cassandraClusterBuilderMaker.create().addContactPoint(node).build();
    Session session = cluster.connect(keyspace);

    List<String> commands = Arrays.asList(new String(commandBuffer,"UTF-8").split(";"));
    for(String command : commands){
        String commandLine = command.trim().replaceAll("^-- .*","");
        if(!commandLine.isEmpty()){
            command = commandLine + ";";
            LOG.info("Execute:\n" + command);
            try {
                session.execute(command);
            } catch (SyntaxError e) {
                LOG.error("Command Failed with " + e.getMessage());
                throw e;
            }
        }
    }

}

@Test
public void testShouldReturnSyntaxError() throws Exception {
  String message = "pc load letter";
  server.prime(when(query).then(SyntaxError(message)));

  thrown.expect(SyntaxError.class);
  thrown.expectMessage(message);
  query();
}

@Test
public void testShouldReturnSyntaxErrorOnPrepare() throws Exception {
  String message = "this Syntax is no good";
  server.prime(when(query).then(SyntaxError(message)).applyToPrepare());

  thrown.expect(SyntaxError.class);
  thrown.expectMessage(message);

  prepare();
}

@Test
public void testShouldNotReturnSyntaxErrorOnPrepare() throws Exception {
  String message = "this Syntax is no good";
  server.prime(when(query).then(SyntaxError(message)).ignoreOnPrepare());

  // should not throw error here.
  prepare();

  thrown.expect(SyntaxError.class);
  thrown.expectMessage(message);

  prepareAndQuery();
}

@Test
public void testShouldNotReturnSyntaxErrorOnPrepareByDefault() throws Exception {
  String message = "this Syntax is no good";
  server.prime(when(query).then(SyntaxError(message)));

  // should not throw error here.
  prepare();

  thrown.expect(SyntaxError.class);
  thrown.expectMessage(message);

  prepareAndQuery();
}

@Test
public void prepareAndExecuteWithCustomExpressions() throws Throwable
{
    session.execute(dropKsstatement);
    session.execute(createKsstatement);
    String table = "custom_expr_test";
    String index = "custom_index";

    session.execute(String.format("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY,cid int,val text);",KEYSPACE,table));
    session.execute(String.format("CREATE CUSTOM INDEX %s ON %s.%s(val) USING '%s'",index,table,StubIndex.class.getName()));
    session.execute(String.format("INSERT INTO %s.%s(id,cid,val) VALUES (0,'test')",table));

    PreparedStatement prepared1 = session.prepare(String.format("SELECT * FROM %s.%s WHERE expr(%s,'foo')",index));
    assertEquals(1,session.execute(prepared1.bind()).all().size());

    PreparedStatement prepared2 = session.prepare(String.format("SELECT * FROM %s.%s WHERE expr(%s,?)",session.execute(prepared2.bind("foo bar baz")).all().size());

    try
    {
        session.prepare(String.format("SELECT * FROM %s.%s WHERE expr(?,'foo bar baz')",table));
        fail("Expected Syntax exception,but none was thrown");
    }
    catch(SyntaxError e)
    {
        assertEquals("Bind variables cannot be used for index names",e.getMessage());
    }
}