This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import abc | |
import functools | |
import json | |
import time | |
import uuid | |
from dataclasses import asdict | |
from dataclasses import dataclass | |
from logging import INFO | |
from logging import Logger | |
from logging import basicConfig |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from typing import Callable | |
from numpy import array | |
from numpy import ndarray | |
def dimension(*expected: int): | |
def wrap_up(func: Callable) -> Callable: | |
def wrapped(*args, **kwargs) -> ndarray: | |
result: ndarray = func(*args, **kwargs) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import tensorflow as tf | |
class VarLenSeries: | |
""" | |
sess = tf.InteractiveSession() | |
series = tf.constant(list(range(5))) | |
sess.run(VarLenSeries(series).to_skip_gram(3)) | |
# # it returns below |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import queue | |
import threading | |
from typing import Callable, Generator, Any | |
_TERM = threading.Event() | |
class Worker(threading.Thread): | |
def __init__(self, source: queue.Queue, destination: queue.Queue, batch_size: int, | |
callback: Callable[[Generator], Any]): |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import redis | |
import time | |
from flask import jsonify, Flask | |
class EventCounter: | |
def __init__(self, seconds: float): | |
pool: redis.ConnectionPool = redis.ConnectionPool(host='localhost', port=6379, db=0) | |
self._r: redis.StrictRedis = redis.StrictRedis(connection_pool=pool) | |
self._s: float = seconds | |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import threading | |
import contextlib | |
class LeakObject: | |
def __init__(self, lifetime: int, cls: object, **kwargs): | |
self._counter = 0 | |
self.lifetime = lifetime | |
self._kwargs = kwargs |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import threading | |
import tracemalloc | |
import time | |
tracemalloc.start() | |
class Sub(threading.Thread): | |
def run(self): | |
while True: | |
time.sleep(3) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import time | |
from typing import Callable, Any | |
def print_timestamp(func: Callable) -> Callable: | |
def wrap_up(**kwargs) -> Any: | |
t0 = time.time() | |
res = func(**kwargs) | |
t1 = time.time() | |
logger.debug({'elapsed_time': t1-t0}) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def retry_if_fail(callback: Callable): | |
""" | |
bigtable sdk auto-reconnect works only if failed. | |
:param callback: some callback function | |
:return: | |
""" | |
def wrap(*args, **kwargs): | |
try: | |
return callback(*args, **kwargs) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from typing import List, Callable, TypeVar | |
from functools import reduce, lru_cache | |
T = TypeVar('T') | |
class MyCollection(List[T]): | |
def reduce(self, f: Callable[[T, T], T]): | |
return reduce(f, self) | |
# example usage | |
obj = MyCollection(range(5)) | |
obj.reduce(lambda x, y: x + y) |
NewerOlder