107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
import requests
|
||
from typing import Optional
|
||
|
||
class BuildIncrementerClient:
|
||
"""
|
||
Клиент для взаимодействия с сервисом счетчиков на базе FastAPI.
|
||
|
||
Позволяет инкрементировать, получать и сбрасывать счетчики с поддержкой namespace,
|
||
application и version.
|
||
|
||
Пример использования:
|
||
client = BuildIncrementerClient(base_url="http://localhost:8000")
|
||
new_value = client.increment("my_namespace", "my_app", "1.0")
|
||
current_value = client.get_value("my_namespace", "my_app", "1.0")
|
||
reset_value = client.reset("my_namespace", "my_app", "1.0")
|
||
"""
|
||
|
||
def __init__(self, base_url: str):
|
||
"""
|
||
Инициализирует клиент с базовым URL сервиса.
|
||
|
||
Args:
|
||
base_url (str): Базовый URL сервиса, например, "http://localhost:8000"
|
||
"""
|
||
self.base_url = base_url.rstrip('/')
|
||
|
||
def increment(self, namespace: str, application: str, version: str) -> int:
|
||
"""
|
||
Инкрементирует счетчик и возвращает новое значение.
|
||
|
||
Args:
|
||
namespace (str): Пространство имен.
|
||
application (str): Имя приложения.
|
||
version (str): Версия приложения.
|
||
|
||
Returns:
|
||
int: Новое значение счетчика.
|
||
|
||
Raises:
|
||
requests.HTTPError: Если запрос завершился с ошибкой.
|
||
"""
|
||
url = f"{self.base_url}/{namespace}/{application}/{version}"
|
||
response = requests.post(url)
|
||
self._handle_response(response)
|
||
return response.json()['value']
|
||
|
||
def get_value(self, namespace: str, application: str, version: str) -> int:
|
||
"""
|
||
Получает текущее значение счетчика.
|
||
|
||
Args:
|
||
namespace (str): Пространство имен.
|
||
application (str): Имя приложения.
|
||
version (str): Версия приложения.
|
||
|
||
Returns:
|
||
int: Текущее значение счетчика.
|
||
|
||
Raises:
|
||
requests.HTTPError: Если запрос завершился с ошибкой.
|
||
"""
|
||
url = f"{self.base_url}/{namespace}/{application}/{version}"
|
||
response = requests.get(url)
|
||
self._handle_response(response)
|
||
return response.json()['value']
|
||
|
||
def reset(self, namespace: str, application: str, version: str) -> int:
|
||
"""
|
||
Сбрасывает счетчик до 0 и возвращает новое значение.
|
||
|
||
Args:
|
||
namespace (str): Пространство имен.
|
||
application (str): Имя приложения.
|
||
version (str): Версия приложения.
|
||
|
||
Returns:
|
||
int: Новое значение счетчика после сброса.
|
||
|
||
Raises:
|
||
requests.HTTPError: Если запрос завершился с ошибкой.
|
||
"""
|
||
url = f"{self.base_url}/{namespace}/{application}/{version}"
|
||
response = requests.delete(url)
|
||
self._handle_response(response)
|
||
return response.json()['value']
|
||
|
||
def _handle_response(self, response: requests.Response):
|
||
"""
|
||
Обрабатывает ответ от сервера, выбрасывая исключение при ошибке.
|
||
|
||
Args:
|
||
response (requests.Response): Ответ от сервера.
|
||
|
||
Raises:
|
||
requests.HTTPError: Если ответ содержит ошибку.
|
||
"""
|
||
try:
|
||
response.raise_for_status()
|
||
except requests.HTTPError as e:
|
||
try:
|
||
error_detail = response.json().get('detail', '')
|
||
raise requests.HTTPError(f"{e}, Detail: {error_detail}") from None
|
||
except ValueError:
|
||
# Если ответ не JSON, просто выбросить исходную ошибку
|
||
raise
|
||
|