whchi

from functools import partial
from multiprocessing import Pool, cpu_count


def batch(iterable, n=1):
    iter_len = len(iterable)
    for ndx in range(0, iter_len, n):
        yield iterable[ndx:min(ndx + n, iter_len)]

whchi

import os

def get_file_list(src_dir):
    input_path = []
    for dir_path, subdir, filenames in os.walk(src_dir):
        for f in filenames:
            input_path.append(os.path.join(dir_path, f))

    return input_path

whchi

version: '3.8'
services:
  ### Redis Master ################################################
  redis-master:
    image: redis
    restart: unless-stopped
    ports:
      - 6381:6379
    networks:
      local_network:

whchi

from fastapi import FastAPI
from my_middleware import MyMiddleware

app = FastAPI()
app.add_middleware(JsonResponseFormatter)

whchi

from fastapi.exceptions import RequestValidationError
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from app.routes import my_route
app = FastAPI()

# 包覆預設的 422 exception
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request,
                                       exc: RequestValidationError):

whchi

from functools import lru_cache
import os
from dotenv import load_dotenv
from pathlib import Path
from typing import Any

BASE_DIR = Path(__file__).resolve().parent.parent
load_dotenv(dotenv_path=f'{BASE_DIR}/app/.env')

whchi

# coding=utf-8

from html.parser import HTMLParser
from html import unescape
import unicodedata
import re


class MLStripper(HTMLParser):
    """

whchi

import time


def retry(exceptions: tuple, times=5, interval=10):
    def wrapper(f):
        def do(*args, **kwargs):
            attempt = 0
            while attempt < times:
                try:
                    return f(*args, **kwargs)

whchi

def singleton(class_):
    instances = {}

    def get_instance(*args, **kwargs):
        if class_ not in instances:
            instances[class_] = class_(*args, **kwargs)
        return instances[class_]

    return get_instance

whchi

version: '3.9'
services:
  influxdb:
    image: quay.io/influxdb/influxdb:v2.0.2
    ports:
      - 8086:8086
########################################### solid volume
    volumes:
      - ~/docker/mount/var/lib/influxdb:/var/lib/influxdb
    command: influxd run --bolt-path /var/lib/influxdb/influxd.bolt --engine-path /var/lib/influxdb/engine