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

fix
All checks were successful
Deploy Prod / Build (pull_request) Successful in 6s
Details
Deploy Prod / Push (pull_request) Successful in 23s
Details
Deploy Prod / Deploy prod (pull_request) Successful in 23s
Details

Browse Source
This commit is contained in:
Egor Matveev 2024-12-28 14:08:40 +03:00
parent 8c238e7023
commit 104ac7377c
2 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@ -8,4 +8,6 @@ RUN pip install -r requirements.txt
COPY . . COPY . .
ENV PYTHONUNBUFFERED 1
ENTRYPOINT ["python", "main.py"] ENTRYPOINT ["python", "main.py"]

10
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
@ -9,7 +7,6 @@ from app.storage.mongo import tasks
DEFAULT_RETRY_AFTER = 0.2 DEFAULT_RETRY_AFTER = 0.2
locks = collections.defaultdict(asyncio.Lock)
router = fastapi.APIRouter() router = fastapi.APIRouter()
@ -23,10 +20,13 @@ class Response(pydantic.BaseModel):
task: Task|None task: Task|None
@router.get('/api/v1/take', responses={500: {'detail': 'Internal server error'}}) @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)
except Exception as e:
print('GOT ERROR', e)
return Response(task=None)
if not task: if not task:
return Response(task=None) return Response(task=None)
# retry_after_config = configurator.get_config('retry_after') # retry_after_config = configurator.get_config('retry_after')