joblib 多核多线程任务，并统一返回结果集

import time

from joblib import Parallel, delayed


def cpu_intensive_task(n, i):
    print(f"Task {n} startedP{i}")
    start = time.time()
    # 假设是一个耗时的计算任务
    result = sum(i * i for i in range(10 ** 10))
    end = time.time()
    print(f"Task {n} finished in {end - start:.2f} seconds")
    return result


tasks = range(4)  # 假设有4个任务,根据CPU数据限制
results = Parallel(n_jobs=-1)(delayed(cpu_intensive_task)(n, 3) for n in tasks)
print(f"All tasks completed.{results}")

aiomysql

基于pymysql 的异步数据库

pip install colorlog

Flask中使用

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,

    'filters': {
        'exclude_internal': {
            '()': CustomFilter,
        }
    },
    'formatters': {
        'default': {
            '()': CustomFormatter,
            'format': LOGGING_DEF_FORMAT,
        },
        'colorFormat': {
            '()': color.ColoredFormatter,
            'format': f'%(log_color)s{LOGGING_DEF_FORMAT}',
        }
    },
    'handlers': {
        'wsgi': {
            'class': 'logging.StreamHandler',
            'stream': 'ext://flask.logging.wsgi_errors_stream',
            'formatter': 'colorFormat',
            'filters': ['exclude_internal']
        },
        "file": {
            # 'class': '..core.utils.loggingHandler.TimeLoggerRolloverHandler',
            "()": TimeLoggerRolloverHandler,
            'level': 'DEBUG',
            'filename': LOG_FILE_PATH + "<FILENAME>.<SUFFIX>",
            'formatter': 'default',
            'filters': ['exclude_internal']
        },
        "errorFile": {
            # 'class': '..core.utils.loggingHandler.TimeLoggerRolloverHandler',
            "()": TimeLoggerRolloverHandler,
            'level': 'ERROR',
            'filename': LOG_FILE_PATH + "<FILENAME>.error.<SUFFIX>",
            'formatter': 'default',
            # 'filters': ['error_log']
        },
        "sqlalchemy": {
            "()": TimeLoggerRolloverHandler,
            'level': 'DEBUG',
            'filename': LOG_FILE_PATH + "sqlalchemy.log",
            'formatter': 'default'
        }
    },
    'loggers': {
        'sqlalchemy.engine.Engine': {
            'level': 'INFO',
            'handlers': ['wsgi', 'sqlalchemy'],
            'propagate': False  # 不记录到root日志
        }
    },
    'root': {
        'level': 'DEBUG',
        'handlers': ['wsgi', 'file', "errorFile"],
    }
}

apscheduler 定时任务

colorlog 日志多彩输出

sqlalchemy 数据库模型

JWT 不用多说

flask[async] flask异步接口

numpy 科学计算模块，功能太多

import numpy as np

def split_array(arr, k):
    return np.array_split(arr, k)

# 示例
arr = np.arange(10)
k = 3
subarrays = split_array(arr, k)
print(subarrays)