fix
This commit is contained in:
@@ -1,45 +0,0 @@
|
||||
import io
|
||||
|
||||
from minio import Minio
|
||||
from minio.error import MinioException
|
||||
|
||||
import settings
|
||||
|
||||
|
||||
class Client:
|
||||
|
||||
def __init__(self, host: str, access_key: str, secret_key: str, bucket_name: str):
|
||||
self.bucket_name = bucket_name
|
||||
self.cli = Minio(
|
||||
host,
|
||||
access_key=access_key,
|
||||
secret_key=secret_key,
|
||||
secure=False
|
||||
)
|
||||
try:
|
||||
self.cli.make_bucket(bucket_name)
|
||||
except MinioException:
|
||||
pass
|
||||
|
||||
def put_object(self, name: str, data: bytes):
|
||||
self.cli.put_object(self.bucket_name, name, io.BytesIO(data), len(data))
|
||||
|
||||
def get_object(self, name: str) -> bytes:
|
||||
try:
|
||||
return self.cli.get_object(self.bucket_name, name).data
|
||||
except MinioException:
|
||||
return b""
|
||||
|
||||
def delete_object(self, name: str):
|
||||
try:
|
||||
self.cli.remove_object(self.bucket_name, name)
|
||||
except MinioException:
|
||||
pass
|
||||
|
||||
|
||||
minio_client = Client(
|
||||
settings.MINIO_HOST,
|
||||
settings.MINIO_ACCESS_KEY,
|
||||
settings.MINIO_SECRET_KEY,
|
||||
settings.MINIO_BUCKET_NAME
|
||||
)
|
||||
52
tools/queues.py
Normal file
52
tools/queues.py
Normal file
@@ -0,0 +1,52 @@
|
||||
import os
|
||||
import requests
|
||||
import time
|
||||
|
||||
|
||||
stage = os.getenv("STAGE", 'local')
|
||||
if stage == 'local':
|
||||
QUEUES_URL = 'http://localhost:1239'
|
||||
else:
|
||||
QUEUES_URL = 'http://queues:1239'
|
||||
|
||||
|
||||
class QueuesException(Exception):
|
||||
...
|
||||
|
||||
|
||||
class TasksHandlerMixin:
|
||||
def poll(self):
|
||||
while True:
|
||||
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name}).json()
|
||||
task = response.get('task')
|
||||
if not task:
|
||||
time.sleep(0.2)
|
||||
continue
|
||||
try:
|
||||
self.process(task['payload'])
|
||||
except Exception as exc:
|
||||
print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
|
||||
continue
|
||||
try:
|
||||
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
|
||||
if resp.status_code != 202:
|
||||
raise QueuesException
|
||||
except:
|
||||
print(f'Failed to finish task id={task["id"]}')
|
||||
|
||||
@property
|
||||
def queue_name(self):
|
||||
raise NotImplemented
|
||||
|
||||
def process(self, payload):
|
||||
raise NotImplemented
|
||||
|
||||
|
||||
def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int|None = None):
|
||||
resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name}, json={
|
||||
'payload': payload,
|
||||
'seconds_to_execute': seconds_to_execute,
|
||||
'delay': delay,
|
||||
})
|
||||
if resp.status_code != 202:
|
||||
raise QueuesException
|
||||
@@ -1,30 +0,0 @@
|
||||
import contextlib
|
||||
|
||||
import redis
|
||||
|
||||
import settings
|
||||
|
||||
|
||||
class RedisClient:
|
||||
|
||||
def __init__(self, host, password=None):
|
||||
kwargs = {
|
||||
"host": host,
|
||||
}
|
||||
if password:
|
||||
kwargs['password'] = password
|
||||
self.cli = redis.Redis(**kwargs)
|
||||
|
||||
def get(self, key):
|
||||
with self.cli as cli:
|
||||
return cli.get(f"ruletka_{key}")
|
||||
|
||||
def set(self, key, value):
|
||||
with self.cli as cli:
|
||||
cli.set(f"ruletka_{key}", value)
|
||||
|
||||
|
||||
redis_client = RedisClient(
|
||||
settings.REDIS_HOST,
|
||||
settings.REDIS_PASSWORD
|
||||
)
|
||||
@@ -1,98 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
import typing
|
||||
import urllib.parse
|
||||
from threading import Thread
|
||||
from time import sleep
|
||||
|
||||
from requests import get
|
||||
|
||||
|
||||
class PlatformClient:
|
||||
def __init__(self, platform_security_token: str, app_name: str, stage: str, need_poll: bool = True):
|
||||
self.platform_security_token = platform_security_token
|
||||
self.app_name = app_name
|
||||
self.stage = stage
|
||||
self.endpoint = 'https://platform.sprinthub.ru/'
|
||||
self.configs_url = urllib.parse.urljoin(self.endpoint, 'configs/get')
|
||||
self.experiments_url = urllib.parse.urljoin(self.endpoint, 'experiments/get')
|
||||
self.staff_url = urllib.parse.urljoin(self.endpoint, 'is_staff')
|
||||
self.fetch_url = urllib.parse.urljoin(self.endpoint, '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,
|
||||
headers={'X-Security-Token': self.platform_security_token},
|
||||
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]
|
||||
|
||||
|
||||
platform = PlatformClient(
|
||||
os.getenv('PLATFORM_SECURITY_TOKEN'),
|
||||
'Ruletka',
|
||||
os.getenv('STAGE', 'local')
|
||||
)
|
||||
Reference in New Issue
Block a user