feat(proxy/utils.py): allow budget duration in months

Closes https://github.com/BerriAI/litellm/issues/4042
This commit is contained in:
Krrish Dholakia 2024-06-13 16:52:17 -07:00
parent d210eccb79
commit f65752c18b
2 changed files with 75 additions and 3 deletions

View file

@ -1,4 +1,4 @@
from typing import Optional, List, Any, Literal, Union, TYPE_CHECKING from typing import Optional, List, Any, Literal, Union, TYPE_CHECKING, Tuple
import os import os
import subprocess import subprocess
import hashlib import hashlib
@ -2093,14 +2093,32 @@ def get_logging_payload(
raise e raise e
def _duration_in_seconds(duration: str): def _extract_from_regex(duration: str) -> Tuple[int, str]:
match = re.match(r"(\d+)([smhd]?)", duration) match = re.match(r"(\d+)(mo|[smhd]?)", duration)
if not match: if not match:
raise ValueError("Invalid duration format") raise ValueError("Invalid duration format")
value, unit = match.groups() value, unit = match.groups()
value = int(value) value = int(value)
return value, unit
def _duration_in_seconds(duration: str) -> int:
"""
Parameters:
- duration:
- "<number>s" - seconds
- "<number>m" - minutes
- "<number>h" - hours
- "<number>d" - days
- "<number>mo" - months
Returns time in seconds till when budget needs to be reset
"""
value, unit = _extract_from_regex(duration=duration)
if unit == "s": if unit == "s":
return value return value
elif unit == "m": elif unit == "m":
@ -2109,6 +2127,22 @@ def _duration_in_seconds(duration: str):
return value * 3600 return value * 3600
elif unit == "d": elif unit == "d":
return value * 86400 return value * 86400
elif unit == "mo":
now = time.time()
current_time = datetime.fromtimestamp(now)
# Calculate the first day of the next month
if current_time.month == 12:
next_month = datetime(year=current_time.year + 1, month=1, day=1)
else:
next_month = datetime(
year=current_time.year, month=current_time.month + value, day=1
)
# Calculate the duration until the first day of the next month
duration_until_next_month = next_month - current_time
return int(duration_until_next_month.total_seconds())
else: else:
raise ValueError("Unsupported duration unit") raise ValueError("Unsupported duration unit")

View file

@ -26,6 +26,7 @@ from litellm.utils import (
get_max_tokens, get_max_tokens,
get_supported_openai_params, get_supported_openai_params,
) )
from litellm.proxy.utils import _duration_in_seconds, _extract_from_regex
# Assuming your trim_messages, shorten_message_to_fit_limit, and get_token_count functions are all in a module named 'message_utils' # Assuming your trim_messages, shorten_message_to_fit_limit, and get_token_count functions are all in a module named 'message_utils'
@ -445,3 +446,40 @@ def test_redact_msgs_from_logs():
litellm.turn_off_message_logging = False litellm.turn_off_message_logging = False
print("Test passed") print("Test passed")
@pytest.mark.parametrize(
"duration, unit",
[("7s", "s"), ("7m", "m"), ("7h", "h"), ("7d", "d"), ("7mo", "mo")],
)
def test_extract_from_regex(duration, unit):
value, _unit = _extract_from_regex(duration=duration)
assert value == 7
assert _unit == unit
def test_duration_in_seconds():
"""
Test if duration int is correctly calculated for different str
"""
import time
now = time.time()
current_time = datetime.fromtimestamp(now)
print("current_time={}".format(current_time))
# Calculate the first day of the next month
if current_time.month == 12:
next_month = datetime(year=current_time.year + 1, month=1, day=1)
else:
next_month = datetime(
year=current_time.year, month=current_time.month + 1, day=1
)
print("next_month={}".format(next_month))
# Calculate the duration until the first day of the next month
duration_until_next_month = next_month - current_time
expected_duration = int(duration_until_next_month.total_seconds())
value = _duration_in_seconds(duration="1mo")
assert value - expected_duration < 2