mypy type annotations for zerver/lib/utils

This commit is contained in:
David Adamec
2016-06-03 12:39:57 -04:00
committed by Tim Abbott
parent a0b332d6cf
commit 8ad20e9775

View File

@@ -2,6 +2,8 @@
from __future__ import absolute_import
from __future__ import division
from typing import Any, Callable, Optional, Sequence, TypeVar
from six import text_type, binary_type
import base64
import hashlib
import os
@@ -10,7 +12,10 @@ from time import sleep
from django.conf import settings
from six.moves import range
T = TypeVar('T')
def statsd_key(val, clean_periods=False):
# type: (Any, bool) -> str
if not isinstance(val, str):
val = str(val)
@@ -30,15 +35,17 @@ class StatsDWrapper(object):
# as our statsd server supports them but supporting
# pystatsd is not released yet
def _our_gauge(self, stat, value, rate=1, delta=False):
# type: (str, float, float, bool) -> str
"""Set a gauge value."""
from django_statsd.clients import statsd
if delta:
value = '%+g|g' % (value,)
value_str = '%+g|g' % (value,)
else:
value = '%g|g' % (value,)
statsd._send(stat, value, rate)
value_str = '%g|g' % (value,)
statsd._send(stat, value_str, rate)
def __getattr__(self, name):
# type: (str) -> Any
# Hand off to statsd if we have it enabled
# otherwise do nothing
if name in ['timer', 'timing', 'incr', 'decr', 'gauge']:
@@ -57,6 +64,7 @@ statsd = StatsDWrapper()
# Runs the callback with slices of all_list of a given batch_size
def run_in_batches(all_list, batch_size, callback, sleep_time = 0, logger = None):
# type: (Sequence[T], int, Callable[[Sequence[T]], None], int, Optional[Callable[[str], None]]) -> None
if len(all_list) == 0:
return
@@ -77,6 +85,7 @@ def run_in_batches(all_list, batch_size, callback, sleep_time = 0, logger = None
sleep(sleep_time)
def make_safe_digest(string, hash_func=hashlib.sha1):
# type: (text_type, Callable[[binary_type], Any]) -> str
"""
return a hex digest of `string`.
"""
@@ -86,6 +95,7 @@ def make_safe_digest(string, hash_func=hashlib.sha1):
def log_statsd_event(name):
# type: (str) -> None
"""
Sends a single event to statsd with the desired name and the current timestamp
@@ -100,4 +110,5 @@ def log_statsd_event(name):
statsd.incr(event_name)
def generate_random_token(length):
# type: (int) -> str
return base64.b16encode(os.urandom(length // 2)).lower()