Передача параметра в Asyncio из модуля в модуль
Мне необходимо передать параметры объекта класса "formatdf", изменяемые в функции "process" в модуль "Tests.restapi_quart" для ответа на GET запрос. Как это сделать? Пробовал использовать Asyncio.Queue и фреймворк contextvars, но так и не разобрался как правильно передавать.
import asyncio
from contextlib import closing
import db_cl, tks_db
from formdf_cl import FormatDF
import Tests.restapi_quart
# contextvars
getinfofromtks = tks_db.TKS()
formatdf = FormatDF()
dbb = db_cl.MyDatabase('mydb.ini')
async def get_some_data():
# Обработка I/O - получение данных из базы
await asyncio.sleep(5)
ans_inc, ans_out = getinfofromtks.getdf()
return ans_inc, ans_out
async def process(ans_inc, ans_out):
# Обработка данных на CPU
await asyncio.sleep(0)
formatdf.leftjoin(ans_inc, ans_out)
formatdf.make_count() # подсчет и сохранение параметров в объекте класса
async def launcher_main():
while True:
ans_inc, ans_out = await get_some_data() # Блокирующий вызов - ожидание поступления новой порции данных
asyncio.ensure_future(process(ans_inc, ans_out)) # Неблокирующий вызов - обработка данных на CPU
async def main():
await asyncio.gather(launcher_main(),Tests.restapi_quart.app.run_task(debug=True, host='0.0.0.0', port=5000))
if __name__ == "__main__":
with closing(asyncio.get_event_loop()) as event_loop:
event_loop.run_until_complete(main())
Модуль "Tests.restapi_quart":
import quart
from quart import request
import json
import time
import db_cl
dbb = db_cl.MyDatabase('mydb.ini')
app = quart.Quart(__name__)
def resp(code, data):
return quart.Response(
status=code,
mimetype="application/json",
response=to_json(data)
)
def resp_json(code, data):
return quart.Response(
status=code,
mimetype="application/json",
response=data
)
def to_json(data):
return json.dumps(data) + "\n"
@app.route('/api/status2', methods=['GET'])
def get_status():
timestamp = request.args.get("timestamp")
timenow = request.args.get("timenow")
lastone = request.args.get("lastone")
button = request.args.get("button")
if timestamp:
return resp_json(200, dbb.fetch_all_as_df("select * from statistic where timestamp >= " + "'" +
time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(float(timestamp)))
+ "';").to_json())
elif timenow:
return resp(200, {"time is:": time.time()})
elif button:
return resp(200, {"ok": **передать параметр сюда**})
elif lastone:
return resp_json(200, dbb.fetch_all_as_df('select * from statistic ORDER BY timestamp DESC LIMIT 1').to_json())
else:
return resp(200, {"", "ask me about time"})
Ответы (1 шт):
Автор решения: eri
→ Ссылка
У Вас всё в одном треде, так что прицепите formatdf к классу приложения
async def main():
Tests.restapi_quart.app.config["formatdf"] = formatdf
await asyncio.gather(launcher_main(),Tests.restapi_quart.app.run_task(debug=True, host='0.0.0.0', port=5000))
И смотрите атрибуты
elif button:
return resp(200, {"ok": app.config["formatdf"].attr})