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

Compare commits

base: self:753e85b7f2ced921b41e4e99a8fcbf151a44173d
Branches Tags
self:master
self:dev
self:prod
..
compare: self:d1c5b26d1d426ae1b8c18746739f490efa9ea647
Branches Tags
self:dev
self:prod
self:master

No commits in common. "753e85b7f2ced921b41e4e99a8fcbf151a44173d" and "d1c5b26d1d426ae1b8c18746739f490efa9ea647" have entirely different histories.
753e85b7f2 ... d1c5b26d1d

1 changed files with 8 additions and 7 deletions
Show Stats Expand all files Collapse all files

11
app/routers/take.py
View File

@ -1,5 +1,3 @@
import asyncio
import collections
import fastapi import fastapi
import pydantic import pydantic
import typing import typing
@ -7,7 +5,7 @@ import typing
from app.storage.mongo import tasks from app.storage.mongo import tasks
locks = collections.defaultdict(asyncio.Lock) DEFAULT_RETRY_AFTER = 0.2
router = fastapi.APIRouter() router = fastapi.APIRouter()
@ -21,10 +19,13 @@ class Response(pydantic.BaseModel):
task: Task|None task: Task|None
@router.get('/api/v1/take', responses={404: {'description': 'Not found'}}) @router.get('/api/v1/take')
async def execute(queue: typing.Annotated[str, fastapi.Header()]) -> Response: async def execute(queue: typing.Annotated[str, fastapi.Header()]) -> Response:
async with locks[queue]: try:
task = await tasks.take_task(queue) task = await tasks.take_task(queue)
if not task: if not task:
return Response(task=None) return Response(task=None)
return Response(task=Task(id=str(task._id), attempt=task.attempts, payload=task.payload)) return Response(task=Task(id=str(task._id), attempt=task.attempts, payload=task.payload))
except Exception as e:
print("GOT ERROR", e)
return Response(task=None)