1

背景

我正在尝试实现APScheduler github 存储库中提供的RPyC概念验证示例，以便使用 gunicorn 与多个工作人员一起部署我的应用程序（ APScheduler 的常见问题解答部分中指出的问题）。此外，我正在尝试使用Flask-APScheduler这样做，以便能够轻松地从任务功能中工作。app_contexts

问题

在调用通过 RPyC 公开的调度程序服务时，我似乎无法正确提供参数。args更具体地说，在传递参数和kwargs（文字变量）以及存储在暴露的 RPyC 函数的*args和中的变量时，这似乎成为一个问题。**kwargs

基本上，我通常在直接调用时使用的参数scheduler.add_job()在通过 RPyC 路由时不起作用，而是在尝试将 RPyC 公开方法接收到的参数传递给底层调度程序实例时导致如下错误。我怎样才能解决这个问题？

最小的工作示例

python app.py在一个终端和python scheduler.py另一个终端中运行

# app.py
from flask import Flask, current_app, jsonify
from flask_apscheduler import APScheduler
import rpyc

scheduler = APScheduler()

def create_app():
    app = Flask(__name__)
    app.scheduler = rpyc.connect("localhost", 12345)
    app.scheduler = app.scheduler.root  # just so current_app.scheduler can be used like normal

    @app.route("/add_job/<report_id>", methods=["GET"])
    def add_job(report_id):
        """
        This works as expected when using 
        from app import scheduler
        scheduler.add_job(...)
        """
        current_app.scheduler.add_job(
            func="app.tasks:run_report",
            args=(report_id,),
            kwargs={"email_results": True},
            executor="threadpool",
            trigger="cron",
            day="*/1",
            id="reconcile_accounts"
        )
        return jsonify({"status": "scheduled"})

    return app

if __name__ == "__main__":
    app = create_app()
    app.run(debug=True)

# scheduler.py
from rpyc.utils.server import ThreadedServer
import rpyc

from app import create_app, scheduler

class SchedulerService(rpyc.Service):
    def __init__(self):
        self._app = None
        self._scheduler = None

    def on_connect(self, conn):
        # code that runs when a connection is created
        # (to init the service, if needed)
        self._app = create_app()
        self._scheduler = scheduler

    def exposed_add_job(self, func, *args, **kwargs):
        # Problem occurs below when sending *args and **kwargs to Flask-APScheduler, which sends them to APScheduler
        job_id = kwargs.pop("id", None)
        return self._scheduler.add_job(job_id, func, *args, **kwargs)

if __name__ == "__main__":
    server = ThreadedServer(SchedulerService, port=12345, protocol_config={"allow_public_attrs": True})
    try:
        server.start()
    except (KeyboardInterrupt, SystemExit):
        pass
    finally:
        scheduler.shutdown()

追踪簿来自self._scheduler.add_job(job_id, func, *args, **kwargs)

127.0.0.1 - - [08/Jul/2021 10:29:43] "GET /reports/2/run HTTP/1.1" 500 -
Traceback (most recent call last):
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask\app.py", line 2088, in __call__
    return self.wsgi_app(environ, start_response)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask\app.py", line 2073, in wsgi_app
    response = self.handle_exception(e)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask\app.py", line 2070, in wsgi_app
    response = self.full_dispatch_request()
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask\app.py", line 1515, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask\app.py", line 1513, in full_dispatch_request
    rv = self.dispatch_request()
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask\app.py", line 1499, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
  File "C:\Users\mhill\PycharmProjects\reporting\app\views.py", line 189, in run_report
    kwargs={"config": json.dumps(report.serialize())}
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\rpyc\core\netref.py", line 240, in __call__
    return syncreq(_self, consts.HANDLE_CALL, args, kwargs)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\rpyc\core\netref.py", line 63, in syncreq
    return conn.sync_request(handler, proxy, *args)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\rpyc\core\protocol.py", line 473, in sync_request
    return self.async_request(handler, *args, timeout=timeout).value
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\rpyc\core\async_.py", line 102, in value
    raise self._obj
_get_exception_class.<locals>.Derived: dictionary update sequence element #0 has length 6; 2 is required

========= Remote Traceback (1) =========
Traceback (most recent call last):
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\rpyc\core\protocol.py", line 324, in _dispatch_request
    res = self._HANDLERS[handler](self, *args)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\rpyc\core\protocol.py", line 592, in _handle_call
    return obj(*args, **dict(kwargs))
  File "C:/Users/mhill/PycharmProjects/reporting/scheduler.py", line 20, in exposed_add_job
    return self._scheduler.add_job(func, *args, **kwargs)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\flask_apscheduler\scheduler.py", line 168, in add_job
    return self._scheduler.add_job(**job_def)
  File "C:\Users\mhill\PycharmProjects\reporting\venv\lib\site-packages\apscheduler\schedulers\base.py", line 429, in add_job
    'kwargs': dict(kwargs) if kwargs is not None else {},
ValueError: dictionary update sequence element #0 has length 6; 2 is required
4

1 回答 1

0

根据github上的this rpyc issueallow_public_attrs ，可以通过在服务器端和客户端都启用来解决映射字典的问题。由于默认情况下，rpyc 不会公开 dict 方法以支持迭代，**kwargs因此基本上无法工作，因为kwargs没有可访问的 dict 方法。

在您的情况下，您只需像这样更改客户端实例：

app.scheduler = rpyc.connect("localhost", 12345, config={ 'allow_public_attrs': True })
于 2021-12-29T07:46:58.863 回答