MCPcopy Create free account
hub / github.com/Roy3838/Observer / increment_usage

Function increment_usage

api/quota_manager.py:122–138  ·  view source on GitHub ↗
(user_id: str, service: str)

Source from the content-addressed store, hash-verified

120 logger.error(f"Failed to send abuse alert: {e}")
121
122async def increment_usage(user_id: str, service: str) -> int:
123 r = await get_redis()
124 ttl = _seconds_until_midnight()
125
126 # Increment rate limit counter (fixed 60s window)
127 rl_key = f"ratelimit:{user_id}"
128 rl_count = await r.incr(rl_key)
129 if rl_count == 1:
130 await r.expire(rl_key, 60)
131
132 # Increment daily quota counter
133 quota_key = f"quota:{user_id}:{service}"
134 new_count = await r.incr(quota_key)
135 if new_count == 1:
136 await r.expire(quota_key, ttl)
137
138 return new_count
139
140async def get_usage_for_service(user_id: str, service: str) -> int:
141 r = await get_redis()

Callers 7

send_smsFunction · 0.90
send_whatsappFunction · 0.90
make_voice_callFunction · 0.90
send_emailFunction · 0.90
send_pushoverFunction · 0.90
send_telegramFunction · 0.90

Calls 2

get_redisFunction · 0.85
_seconds_until_midnightFunction · 0.85

Tested by

no test coverage detected