百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术教程 > 正文

一个 Python 小项目的小结(python小型项目)

toqiye 2024-08-13 11:23 23 浏览 0 评论

段时间临时接手一个 Python 小项目,这个项目实现的类似一个管控平台,其中核心功能是为算法同学提供机器学习模型训练任务的全流程管理,平台后端基于 Flask 框架实现,前端基于 Ant Design Pro 实现。

代码稍微有些乱,所以做了部分代码的重构,在此做点经验小结。

1、并行化或异步化

部分请求处理逻辑,由于比较耗时,故使用线程池来加速,或者使用独立线程异步处理,或者先存储一个中间状态,由后台定时任务来完成实际的处理工作。对于异步处理结果,前端通过轮询来获取。

线程池的使用,主要使用 map 方法:

from multiprocessing.dummy import Pool

input_list = [...]
pool: Pool = Pool(len(input_list))
pool.map(func, input_list)
pool.close()
pool.join()

独立线程异步处理:

import multiprocessing

p = multiprocessing.Process(target=func, args=(...))
p.start()

定时任务,基于 apscheduler 库实现:

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

scheduler.add_join(func, 'interval', seconds=1)

scheduler.start()

因为对于 Python 应用,通常会使用 gunicorn 这种 WSGI HTTP 服务器以多进程启动多个应用实例,提升请求吞吐能力。但是对于定时任务我们希望只有一个实例,对此,如果使用 gunicorn,可以基于它的 preload 机制来实现:

# wsgi.py
import app

if __name__ == "__main__":
    app.run()
# 注意其中的 --preload 参数
gunicorn --workers=4 --preload --log-level=info --access-logfile=access.log -b 0.0.0.0:8080 wsgi:app

preload 机制简单来说,就是 import app 类所在的模块及其依赖的各个模块(import 过程中会执行其中的语句),然后 fork 出多个进程,每个进程都执行 app.run()。

2、实现一些通用方案对异常进行捕获或重试

def exception_try(times: int = 3, sleep_then_try_seconds=None):
    def decorator(f):
        def wrapper(*args, **kwargs):
            count = 0
            exception = None
            while count < times:
                try:
                    return f(*args, **kwargs)
                except Exception as e:
                    exception = e
                    count += 1
                    logging.exception("Try {} times".format(count))
                    if (sleep_then_try_seconds is not None) and count < times:
                        time.sleep(sleep_then_try_seconds)
            raise exception
        return wrapper
    return decorator
@exception_try(times=3, sleep_then_try_seconds=0.5)
def connect(self):
    return pymysql.connect(host=self.host, user=self.user, password=self.password, db=self.db, charset=self.charset)

这个装饰器方法用于实现异常重试,并且可以指定重试的时间间隔,实际使用下来效果较好。而且也不会因为 try...except 导致大块代码缩进。

确保数据库连接关闭(其它类似资源也可以这样实现)

def with_db(db: Connection, exception_callback=None):
    def decorator(f):
        def db_context(*a, **kw):
            try:
                return f(db, *a, **kw)
            except Exception as e:
                logging.exception(str(e))
                if exception_callback is not None:
                    exception_callback(e)
            finally:
                try:
                    db.close()
                except:
                    pass
        return db_context

    return decorator
# 将 conf.db.connect() 对象作为 delete_task_from_job_queue 的第一个参数注入,task_id 这个参数以不定参数的方式传入 delete_task_from_job_queue
with_db(conf.db.connect())(delete_task_from_job_queue)(task_id)

这个装饰器方法用于确保数据库连接在异常发生也能正常关闭,防止资源泄露。

3、循环等待或超时

class TimeoutCondition(object):

    def __init__(self, condition_func, timeout_seconds):
        self.condition = condition_func
        self.timeout = timeout_seconds
        self.begin = None
        self.timeout_false = True
        self.cond_true = True

    def __bool__(self):
        if self.begin is None:
            self.begin = timeit.default_timer()
        self.cond_true = self.condition()
        self.timeout_false = self.timeout <= 0 or (timeit.default_timer() - self.begin) < self.timeout
        return self.cond_true and self.timeout_false

    def is_timeout(self):
        return self.cond_true and not self.timeout_false
cond = TimeoutCondition(lambda : len(service_list) == 0, 5)
while cond:
    time.sleep(1)
    service_list = get_service_list()
if cond.is_timeout():
    return None, None

TimeoutCondition 用于实现循环等待某个条件满足,但为了避免死循环,所以加一个超时条件判断。实例化参数第一个是原始的条件判断 lambda 语句,第二个是一个超时设置。另外,借助魔术方法 __bool__ ,让 TimeoutCondtion 的实例用起来像是一个布尔变量,调用 is_timeout() 方法可以区分循环等待退出是因为原始条件满足,还是超时退出的。

4、按部署环境配置应用的行为

应用在不同的环境(开发、测试、生产)中应该允许加载不同的配置,配置不同的行为。

当前应用处于什么环境,可以通过环境变量来配置,应用初始化时最先检测当前处于什么环境,之后的初始化流程就可以依据环境配置来加载配置,定制应用行为。

# conf/__init__.py
class AppConfig(object):
    app_env = os.getenv('APP_ENV', 'development')
    is_prod = app_env == 'production'
    is_dev = app_env == 'development'
    is_testing = app_env == 'testing'

    # 其余应用配置项
    ...

conf = AppConfig()


def _load_config_by_env(env: str):
    '''
    不同环境加载不同的配置文件
    配置目录结构:
    conf/
        __init__.py
        development.py
        production.py
        testing.py
    '''
    module = importlib.import_module('conf.{}'.format(env))
    if not hasattr(module, 'Config'):
        logging.warning('Not find {} config'.format(env))
        return
    for name, value in getattr(module, 'Config').__dict__.items():
        if name.startswith('__'):
            continue
        conf.__dict__[name] = value
# 根据环境配置日志级别
log_level = logging.INFO if conf.is_prod else logging.DEBUG
logging.basicConfig(format=consts.LOG_FORMAT, level=log_level)

相关推荐

暗网是什么?到底有多可怕?(暗网有多可怕)

https://mp.weixin.qq.com/s/O7l4rveLnXLt-XE2A0WZ3g

“暗网”是什么,到底有多可怕,互联网的“另外一个世界”

来源成戈科技说悄悄告诉你何为“暗网”,暗网的用途是什么?到底多可怕?暗网的英文是“deepnet或deepweb”,也就是深网的意思,我们都知道,南北极的冰川看起来非常雄伟,但是你物理稍微懂一点你就...

什么是暗网?暗网有什么用?进来我告诉你

网的英文是“deepnet或deepweb”,也就是深网的意思,我们都知道,南北极的冰川看起来非常雄伟,但是你物理稍微懂一点你就会知道,我们看到的只是冰山的10%,还有90%在水面以下,很容易看出来,...

网工跳槽必备,2022年最新大厂高频技术面试真题整理

2022年金三银四正在进行,很多粉丝问我要网络工程师面试方面的资料,有在学校准备实习的,有已经工作准备跳槽的。我翻看最近的笔记,正好有整理一份今年大厂的面试题分享给大家。可以先说的是,国内的互联网面试...

三天吃透操作系统面试八股文(三天吃透计算机网络八股文)

操作系统的四个特性?并发:同一段时间内多个程序执行(与并行区分,并行指的是同一时刻有多个事件,多处理器系统可以使程序并行执行)...

掌握前端面试八股文,提升个人能力,实战面试必备!

前言:前端面试是每个前端开发者职业发展中的重要环节。掌握一些常见的前端面试题目,不仅能够在面试中表现出色,还能够提升自身的技术能力和知识广度。本文将为你介绍一些实用的前端面试题目,帮助你在面试中脱颖而...

进大厂必备的Java八股文大全(2022最强精简易懂版)

2022年秋招即将来临,很多同学会问Java面试八股文有必要背吗?答案是,必须背,博主是个三本,今年凭借这篇八股文斩获了多个大厂暑期实习offer,相信秋招一定也可以发挥重要作用。你可以讨厌这种模式,...

2022最新软件测试八股文,能不能拿心仪Offer就看你背得怎样了

前言鉴于目前测试就业越来越严峻,内卷也成了测试领域的代名词了。我的一个HR朋友告诉我,由于门槛较低,现在普通测试岗(偏功能)的投递比已经将近100,也就是一个岗位差不多有百分简历投进来。所以现在还想从...

《面试八股文》之Dubbo17卷(面试 八股文)

作者:moon原文:https://mp.weixin.qq.com/s/-kVf5qWqcw-4AJF7LL3uWw前言...

前端面试八股文?不存在的!(2021前端面试问题)

最近刷到一位前端小姐姐海外工作分享的视频,分享了她的求职,面试以及工作生活的感受,了解到海外求职面试的一个过程,其中我们经常聊的...

凭借这份《2022测试八股文》候选者逆袭面试官,offer拿到手软

《2023测试面试八股文》800道软件测试面试真题,高清打印版打包带走,横扫软件测试面试高频问题,涵盖测试理论、Linux、MySQL、Web测试、接口测试、App测试、Python、Selen...

面试常考八股文及算法(一)(八股文的要求)

define和const的区别1.define是预处理指令,用于创建符号常量。`const`是C和C++的关键字,用于创建具有常量值的变量,本质是只读变量。2.`define`在预处理阶...

面试必备(背)--计算机网络八股文系列

1.OSI七层、TCP/IP四层的关系和区别?七层模型,亦称OSI(OpenSystemInterconnection),它是一个七层的、抽象的模型体,不仅包括一系列抽象的术语或概念,也包括...

Java面试一定要坚持背的八股文!错过会很可惜!

很多人对java面试题都嗤之以鼻,认为无法衡量出一个程序员的真实水平。还有一部分原因,也是因为太难背了。那我们到底还要不要背?背!当然要背!但也不是死记硬背。在我们背诵的过程中,就把java的核心知识...

面试问八股文的公司都是垃圾?(八股文负面影响)

做医生的需要有医师资格证,做财务的有CPA证书,做教师的有教师资格证等等,做程序员的从来没听说过面试的时候需要你提供什么证书,既然没有可以证明从业能力的证书,那面试的时候如何来判断候选人的基本能力呢?...

取消回复欢迎 发表评论: