我改進(jìn)了去年的應(yīng)用程序。一開(kāi)始有兩個(gè)不同的 python 應(yīng)用程序——第一個(gè)用于統(tǒng)計(jì)數(shù)據(jù)，第二個(gè)——帶有 GET 請(qǐng)求的網(wǎng)絡(luò)服務(wù)器 gunicorn+flask。（這兩種服務(wù)都在 centos 中）Statistics 進(jìn)行計(jì)數(shù)并將所有內(nèi)容存儲(chǔ)在 Postgres 中。Web 服務(wù)器連接到該 Postgres 數(shù)據(jù)庫(kù)并響應(yīng) GET 請(qǐng)求。在重寫(xiě)的版本中，我使用 pandas 框架進(jìn)行了所有統(tǒng)計(jì)，現(xiàn)在我想將這兩個(gè)應(yīng)用程序合并為一個(gè)。我使用 asyncio 來(lái)獲取數(shù)據(jù)和計(jì)數(shù)統(tǒng)計(jì)。一切正常，現(xiàn)在我要添加 Web 服務(wù)器以響應(yīng) GET。部分代碼：import asynciofrom contextlib import closingimport db_cl, tks_dbfrom formdf_cl import FormatDFgetinfofromtks = tks_db.TKS() # Class object to connect to third party databaseformatdf = FormatDF() # counting class object, that stores some datadbb = db_cl.MyDatabase('mydb.ini') # Class object to connect to my databaseasync def get_some_data(): # getting information from third party database every 5 seconds. await asyncio.sleep(5) ans_inc, ans_out = getinfofromtks.getdf() return ans_inc, ans_out # two huge dataframes in pandasasync def process(ans_inc, ans_out): # counting data on CPU await asyncio.sleep(0) formatdf.leftjoin(ans_inc, ans_out) # storing statistics in my Database dbb.query_database('INSERT INTO statistic (timestamp, outgoing, incoming, stats) values (%s, %s,%s, %s)', formatdf.make_count()) dbb.commit_query()async def main(): while True: ans_inc, ans_out = await get_some_data() # blocking, get data from third party database asyncio.ensure_future(process(ans_inc, ans_out)) # computingif __name__ == "__main__": with closing(asyncio.get_event_loop()) as event_loop: event_loop.run_until_complete(main())現(xiàn)在我希望將 http 服務(wù)器添加為線程應(yīng)用程序（使用 flask 或 aiohttp），它將使用類(lèi)對(duì)象“formatdf”中的參數(shù)響應(yīng) GET 請(qǐng)求。包含這些功能的最佳方式是什么？
2 回答
胡說(shuō)叔叔
TA貢獻(xiàn)1804條經(jīng)驗(yàn) 獲得超8個(gè)贊
我設(shè)法添加了一個(gè) http 服務(wù)器作為協(xié)程。首先我嘗試使用 aiohttp，但最終我找到了 Quart（與 Flask 相同，但它使用 Asyncio）。在 Quart 上運(yùn)行 http 服務(wù)器的示例代碼：
import quart
from quart import request
import json
import time
app = quart.Quart(__name__)
def resp(code, data):
return quart.Response(
status=code,
mimetype="application/json",
response=to_json(data)
)
def to_json(data):
return json.dumps(data) + "\n"
@app.route('/')
def root():
return quart.redirect('/api/status2')
@app.errorhandler(400)
def page_not_found(e):
return resp(400, {})
@app.errorhandler(404)
def page_not_found(e):
return resp(400, {})
@app.errorhandler(405)
def page_not_found(e):
return resp(405, {})
@app.route('/api/status2', methods=['GET'])
def get_status():
timestamp = request.args.get("timestamp")
delay = request.args.get("delay")
if timestamp:
return resp(200, {"time": time.time()})
elif delay:
return resp(200, {"test is:": '1'})
else:
return resp(200, {"", "ask me about time"})
if __name__ == "__main__":
app.run(debug=True, host='0.0.0.0', port=5000)
為了將此代碼添加為協(xié)同程序，我使用await asyncio.gather()并使用了 app.run_task 而不是 app.run。像這樣更改了我的問(wèn)題中的代碼：
async def launcher_main():
while True:
ans_inc, ans_out = await get_some_data()
asyncio.ensure_future(process(ans_inc, ans_out))
async def main():
await asyncio.gather(launcher_main(),
restapi_quart.app.run_task(debug=True, host='0.0.0.0', port=5000))
剩下的最后一個(gè)問(wèn)題是使“formatdf”類(lèi)對(duì)象的可用參數(shù)到我的 http 服務(wù)器。我已經(jīng)實(shí)現(xiàn)了Tests.restapi_quart.app.config["formatdf"] = formatdf向 process(...) 函數(shù)添加行。從 quart 調(diào)用它：
elif button:
return resp(200, {"ok": app.config["formatdf"].attr})
精慕HU
TA貢獻(xiàn)1845條經(jīng)驗(yàn) 獲得超8個(gè)贊
我只需要為我的應(yīng)用程序解決這個(gè)問(wèn)題。這是在現(xiàn)有異步應(yīng)用程序中運(yùn)行 aiohttp 服務(wù)器的方法。
https://docs.aiohttp.org/en/stable/web_lowlevel.html
import asyncio
from aiohttp import web
async def web_server():
print(f'Configuring Web Server')
app = web.Application()
app.add_routes([web.get('/hello', web_hello)])
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner)
print(f'Starting Web Server')
await site.start()
print(f'Web Server Started')
#run forever and ever
await asyncio.sleep(100*3600)
async def web_hello(request):
return web.Response(text="Hello World")
async def main():
tasks = []
#Web Server
tasks.append(asyncio.create_task(web_server()))
results = await asyncio.gather(*tasks)
if __name__ == '__main__':
asyncio.run(main())
添加回答
舉報(bào)
0/150
提交
取消