self/queues-py3-grpc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix
All checks were successful
Deploy Dev / Build (pull_request) Successful in 8s
Details
Deploy Dev / Push (pull_request) Successful in 8s
Details
Deploy Dev / Deploy dev (pull_request) Successful in 9s
Details

Browse Source
This commit is contained in:
emmatveev 2024-12-13 02:06:21 +03:00
parent 0228bd03fe
commit 7b96aa6315
1 changed files with 1 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
server.py
View File

@ -2,7 +2,6 @@ import asyncio
import datetime import datetime
import grpc import grpc
import bson import bson
import os
from queues import tasks_pb2 from queues import tasks_pb2
from queues import tasks_pb2_grpc from queues import tasks_pb2_grpc
@ -10,18 +9,6 @@ from queues import tasks_pb2_grpc
from utils import time from utils import time
from storage.mongo import tasks from storage.mongo import tasks
from utils import configurator
DEFAULT_RETRY_AFTER = 0.2
client = configurator.Client(
'queues',
os.environ['STAGE'],
need_poll=True,
)
def get_feature(feature_db, point): def get_feature(feature_db, point):
"""Returns Feature at given location or None.""" """Returns Feature at given location or None."""
@ -49,9 +36,7 @@ class TasksServicer(tasks_pb2_grpc.TasksServicer):
task = await tasks.take_task(request.queue) task = await tasks.take_task(request.queue)
if not task: if not task:
return tasks_pb2.TakeResponse(task=None) return tasks_pb2.TakeResponse(task=None)
retry_after_settings = client.get_config('retry_after_settings') return tasks_pb2.TakeResponse(task=tasks_pb2.Task(id=str(task._id), attempt=task.attempts, payload=task.payload))
retry_after = retry_after_settings.get(request.queue) or retry_after_settings.get('__default__') or DEFAULT_RETRY_AFTER
return tasks_pb2.TakeResponse(task=tasks_pb2.Task(id=str(task._id), attempt=task.attempts, payload=task.payload), retry_after=retry_after)
async def Finish(self, request: tasks_pb2.FinishRequest, context) -> tasks_pb2.EmptyResponse: async def Finish(self, request: tasks_pb2.FinishRequest, context) -> tasks_pb2.EmptyResponse:
if await tasks.finish_task(bson.ObjectId(request.id)): if await tasks.finish_task(bson.ObjectId(request.id)):