Python技术迷

公司趁我下班后偷看我电脑,看到我电脑里有信息,git拉的代码说我干私活要把我开了。。

刚看到个贴子,说程序员下班后公司偷看电脑,发现他本地有git拉的代码,就扣帽子说他干私活,要开人,还把1.2万工资扣到只剩4千😓

Image

作为程序员我挺能共情,这种事在一些小公司真的常见:只想压榨员工,不讲契约精神。网友们有的说“别忍,直接仲裁”,有的觉得“认倒霉,下次长记性”。但我觉得核心问题是,公司侵犯了员工的隐私,还恶意赖账,这是职场最恶心的操作。技术人写点代码、调试个项目,本来就是常态,他们却借机扣工资,完全是打压。

换个角度看,这件事也提醒我们:别把重要的私活代码放公司电脑,避免给人借口;遇到这类无良公司,最好留好证据走法律途径,别怕麻烦。说到底,写代码是我们的饭碗,但人格和尊严比工资更重要。【备注:文末可领最新资料】

面试题:有时间限制的 Promise 对象

昨天晚上十一点多,我在公司楼下吹风,手机这边有人问:东哥,Python 里能不能搞个“有时间限制的 Promise”?我当时困得要命但还是回了句,嗯…可以的,只是名字不叫 Promise,叫 Future/Task,不过意思差不多,对吧。

场景你们肯定遇到过:一个异步活儿,可能很慢,比如请求外部接口;可我只想等 300ms,超时就拉倒,别把整体卡死。JS 里常说给 Promise 加超时包装;在 Python 里,咱用 asyncio,核心招数其实就一个——asyncio.wait_for,它会在指定时限内等协程完事儿,不然抛 TimeoutError。思路非常直白,但是…业务里总要顺手点:要不要在超时后返回个默认值?要不要把还在跑的任务取消掉?要不要像 Promise 那样链式用?这些小细节,落地就舒服了。

先上最实用的小函数

我先写了个非常接地气的封装,名字随便起,反正好记就行。

import asyncio
from typing import Any, Awaitable, Optional

asyncdefwith_timeout(coro: Awaitable[Any],
                       timeout: float,
                       default: Optional[Any] = None,
                       cancel_on_timeout: bool = True)
 -> Any:

    task = asyncio.create_task(coro)
try:
returnawait asyncio.wait_for(task, timeout)
except asyncio.TimeoutError:
if cancel_on_timeout:
            task.cancel()
if default isnotNone:
return default
raise

用法就像:“给我这件事等 0.3 秒,等不到就返回 None 别吵我”。比如我在工位上测接口,外侧逻辑只想快点回页面:

asyncdefslow_api():
await asyncio.sleep(1.0)
return {"ok": True}

asyncdefmain():
    data = await with_timeout(slow_api(), 0.3, default={"ok": False, "reason": "timeout"})
    print(data)

asyncio.run(main())

你看,300ms 到点就给默认值,页面不卡,体验也不至于崩。有人会问,东哥,那不取消会怎样?嗯…不取消的话,那个协程还在后台跑,可能白浪费资源,还可能改状态,比较危险,所以我默认是取消,除非你明确说别取消。

想更“像 Promise 一点”的轻量类

有同学执念比较深,非要 new 一个“有期限的 Promise”。行,我就包个小壳,靠 __await__ 让它能 await,看起来就像拿到一个带时限的对象。

import asyncio
from typing import Any, Optional

classTimeoutPromise:
def__init__(self, coro, timeout: float,
                 default: Optional[Any] = None,
                 cancel_on_timeout: bool = True)
:

        self._task = asyncio.create_task(coro)
        self._timeout = timeout
        self._default = default
        self._cancel = cancel_on_timeout

def__await__(self):
return self._await_impl().__await__()

asyncdef_await_impl(self):
try:
returnawait asyncio.wait_for(self._task, self._timeout)
except asyncio.TimeoutError:
if self._cancel:
                self._task.cancel()
if self._default isnotNone:
return self._default
raise

用起来像这样,我在茶水间边倒水边敲的例子,别介意变量名乱七八糟:

asyncdeffetch_user(uid: int):
await asyncio.sleep(0.8)
return {"id": uid, "name": "alice"}

asyncdefmain():
    p = TimeoutPromise(fetch_user(42), timeout=0.5, default={"id": 42, "name": "guest"})
    user = await p
    print(user)

asyncio.run(main())

这玩意儿不搞 then/catch 链式,Python 里 await 就够顺手了。真要链,你也能再包一层,但别折腾过度,维护会累。

一个是取消语义:task.cancel() 会给协程抛 CancelledError,如果你的协程里吞了异常或者做了清理,要记得 finally 里把资源关好,不然内存、连接这些会漏。一个是默认值:给默认值确实让上层舒服,但容易藏 bug,比如你以为真有数据,其实是兜底回来的,所以默认值建议带上原因字段,像我上面那样。还有聚合等待:多个并发任务想“一起等但有总时限”,你可以把 asyncio.gather 外面再套一层 with_timeout,整体体验会更稳。

好了我先去热个咖啡,脑袋有点转不动了,等会儿谁要把它塞到框架中间件里…你们直接把 with_timeout 抽成工具函数就行,别整太花,真没必要。

-END-

我为大家打造了一份RPA教程,完全免费:songshuhezi.com/rpa.html

🔥虎哥私藏精品🔥

虎哥作为一名老码农,整理了全网最全《python高级架构师资料合集》,总量高达650GB,点击下方公众号回复关键字 python 全部免费领