master #40

Merged
emmatveev merged 20 commits from master into dev 2024-12-29 13:32:45 +03:00
7 changed files with 32 additions and 103 deletions

View File

@ -6,7 +6,6 @@ services:
image: mathwave/sprint-repo:queues
networks:
- queues-development
- configurator
environment:
MONGO_HOST: "mongo.develop.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
@ -22,5 +21,3 @@ services:
networks:
queues-development:
external: true
configurator:
external: true

View File

@ -6,7 +6,6 @@ services:
image: mathwave/sprint-repo:queues
networks:
- queues
- configurator
environment:
MONGO_HOST: "mongo.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
@ -22,5 +21,3 @@ services:
networks:
queues:
external: true
configurator:
external: true

View File

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

View File

@ -5,10 +5,8 @@ import pydantic
import typing
from app.storage.mongo import tasks
# from app.utils.configurator import configurator
DEFAULT_RETRY_AFTER = 0.2
locks = collections.defaultdict(asyncio.Lock)
router = fastapi.APIRouter()
@ -21,15 +19,12 @@ class Task(pydantic.BaseModel):
class Response(pydantic.BaseModel):
task: Task|None
retry_after: float
@router.get('/api/v1/take')
@router.get('/api/v1/take', responses={404: {'description': 'Not found'}})
async def execute(queue: typing.Annotated[str, fastapi.Header()]) -> Response:
async with locks[queue]:
task = await tasks.take_task(queue)
if not task:
return Response(task=None)
# retry_after_config = configurator.get_config('retry_after')
# retry_after = retry_after_config.get(queue) or retry_after_config.get('default') or DEFAULT_RETRY_AFTER
return Response(task=Task(id=str(task._id), attempt=task.attempts, payload=task.payload), retry_after=DEFAULT_RETRY_AFTER)
return Response(task=Task(id=str(task._id), attempt=task.attempts, payload=task.payload))

View File

@ -1,88 +0,0 @@
import json
import urllib.parse
import os
from threading import Thread
from time import sleep
from requests import get
class ConfiguratorClient:
def __init__(self, app_name: str, stage: str, need_poll: bool = True):
self.app_name = app_name
self.stage = stage
self.endpoint = 'http://configurator/'
self.fetch_url = urllib.parse.urljoin(self.endpoint, '/api/v1/fetch')
self.config_storage = {}
self.experiment_storage = {}
self.staff_storage = {}
self.poll_data()
if need_poll:
self.poll_data_in_thread()
def poll_data_in_thread(self):
def inner():
while True:
sleep(30)
self.fetch()
Thread(target=inner, daemon=True).start()
def poll_data(self):
self.fetch(with_exception=True)
def request_with_retries(self, url, params, with_exception=False, retries_count=3):
exception_to_throw = None
for _ in range(retries_count):
try:
response = get(
url,
params=params
)
if response.status_code == 200:
return response.json()
print(f'Failed to request {url}, status_code={response.status_code}')
exception_to_throw = Exception('Not 200 status')
except Exception as exc:
print(exc)
exception_to_throw = exc
sleep(1)
print(f'Failed fetching with retries: {url}, {params}')
if with_exception:
raise exception_to_throw
def fetch(self, with_exception=False):
if self.stage == 'local':
local_platform = json.loads(open('local_platform.json', 'r').read())
self.config_storage = local_platform['configs']
self.experiment_storage = local_platform['experiments']
self.staff_storage = {
key: set(value)
for key, value in local_platform['platform_staff'].items()
}
return
response_data = self.request_with_retries(self.fetch_url, {
'project': self.app_name,
'stage': self.stage,
}, with_exception)
self.config_storage = response_data['configs']
self.experiment_storage = response_data['experiments']
self.staff_storage = {
key: set(value)
for key, value in response_data['platform_staff'].items()
}
def is_staff(self, **kwargs):
for key, value in kwargs.items():
if value in self.staff_storage[key]:
return True
return False
def get_config(self, name):
return self.config_storage[name]
def get_experiment(self, name):
return self.experiment_storage[name]
# configurator = ConfiguratorClient('queues', os.getenv('STAGE', 'local'))

10
main.py
View File

@ -8,14 +8,20 @@ from app.routers import finish
from app.storage import mongo
app = fastapi.FastAPI()
app = fastapi.FastAPI(debug=True)
app.include_router(take.router)
app.include_router(put.router)
app.include_router(finish.router)
mongo.create_indexes()
@app.exception_handler(Exception)
async def unicorn_exception_handler(request: fastapi.Request, exc: Exception):
return fastapi.JSONResponse(
status_code=500,
content={"message": f"Oops! {exc.name} did something. There goes a rainbow... {exc}"},
)
mongo.create_indexes()
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=1239)

View File

@ -5,20 +5,40 @@ certifi==2024.12.14
charset-normalizer==3.4.1
click==8.1.7
dnspython==2.7.0
email_validator==2.2.0
fastapi==0.115.4
fastapi-cli==0.0.7
h11==0.14.0
httpcore==1.0.7
httptools==0.6.4
httpx==0.28.1
idna==3.10
Jinja2==3.1.5
markdown-it-py==3.0.0
MarkupSafe==3.0.2
mdurl==0.1.2
motor==3.6.0
pydantic==2.9.2
pydantic_core==2.23.4
Pygments==2.18.0
pymongo==4.9.2
python-dotenv==1.0.1
python-multipart==0.0.20
pytz==2024.2
PyYAML==6.0.2
redis==5.2.0
requests==2.32.3
rich==13.9.4
rich-toolkit==0.12.0
shellingham==1.5.4
six==1.16.0
sniffio==1.3.1
starlette==0.41.2
typer==0.15.1
typing_extensions==4.12.2
tzlocal==5.2
urllib3==2.3.0
uvicorn==0.32.0
uvloop==0.21.0
watchfiles==1.0.3
websockets==14.1