#67 转载:Python 异步编程与数据库
Python SQLAlchemy asyncio 2020-11-22这是大神 zzzeek 2015 年发表的一篇文章,详细介绍了关于 SQLAlchemy 与异步编程的一些事情。解答了我关于如何实现异步编程的一些疑惑。
我曾反复阅读这篇文章好多遍,以求能够更加准确地领会到大佬阐述的意思。我认为每个 Python 的使用者都应该阅读阅读。
coding in a complicated world
这是大神 zzzeek 2015 年发表的一篇文章,详细介绍了关于 SQLAlchemy 与异步编程的一些事情。解答了我关于如何实现异步编程的一些疑惑。
我曾反复阅读这篇文章好多遍,以求能够更加准确地领会到大佬阐述的意思。我认为每个 Python 的使用者都应该阅读阅读。
RQ (Redis Queue) is a simple Python library for queueing jobs and processing them in the background with workers. It is backed by Redis and it is designed to have a low barrier to entry. It can be integrated in your web stack easily.
翻译:RQ (Redis Queue)是一个简单的 Python 库,用于将作业排队并在后台与 worker 一起处理它们。它由 Redis 支持,其设计具有较低的进入门槛。它可以很容易地集成到您的 web 堆栈中。
This project has been inspired by the good parts of Celery, Resque and this snippet, and has been created as a lightweight alternative to existing queueing frameworks, with a low barrier to entry.
启动 worker 进程:
rq worker
rq worker --url redis://:secrets@example.com:1234/9
项目中添加任务:
from redis import Redis
from rq import Queue
# 队列
q = Queue(connection=Redis())
# 添加任务
from my_module import count_words_at_url
result = q.enqueue(count_words_at_url, 'http://nvie.com')
# 关于重试
from rq import Retry
# 失败之后立即重试
queue.enqueue(say_hello, retry=Retry(max=3))
# 失败之后间隔指定时间重试
queue.enqueue(say_hello, retry=Retry(max=3, interval=[10, 30, 60]))
获取结果:
平时删除文件都是 os.unlink 和 os.remove 中随便选一个,今天突然想看看这两个方法有什么不一样。
remove 和 unlink 实际上来自 Modules/posixmodule.c
可以看到这两个方法实际上相同。
/*[clinic input]
os.remove = os.unlink
Remove a file (same as unlink()).
If dir_fd is not None, it should be a file descriptor open to a directory,
and path should be relative; path will then be relative to that directory.
dir_fd may not be implemented on your platform.
If it is unavailable, using it will raise a NotImplementedError.
[clinic start generated code]*/
static PyObject *
os_remove_impl(PyObject *module, path_t *path, int dir_fd)
/*[clinic end generated code: output=a8535b28f0068883 input=e05c5ab55cd30983]*/
{
return os_unlink_impl(module, path, dir_fd);
}
Python 3 的 Path 对象中也有一个 unlink 方法(pathlib.Path.unlink
):
def unlink(self, missing_ok=False):
"""
Remove this file or link.
If the path is a directory, use rmdir() instead.
"""
try:
os.unlink(self)
except FileNotFoundError:
if not missing_ok:
raise
顺便对删除目录做一个整理:
# os.remove(path: StrOrBytesPath, *, dir_fd: int | None = ...)
# os.unlink(path: StrOrBytesPath, *, dir_fd: int | None = ...)
os.mkdir(path: StrOrBytesPath, mode: int = ..., *, dir_fd: int | None = ...)
os.rmdir(path: StrOrBytesPath, *, dir_fd: int | None = ...)
os.makedirs(name: StrOrBytesPath, mode: int = ..., exist_ok: bool = ...)
os.removedirs(name: StrOrBytesPath)
pathlib.Path.rmdir -> os.rmdir
shutil.rmtree(path, ignore_errors=False, onerror=None, *, dir_fd=None)
asyncio
来自 Quora,原文标题:Why is Python so popular despite being so slow?
说明:这里谈的 Python,很大程度上说的是 CPython 这个标准实现,而不是 Python 这门语言。
Tkinter 可以认为是 Python 官方支持的 GUI 框架,接近标准库的地位,样式风格就是极简。
基于 Tcl/Tk。
PyQt Riverbank Computing 提供的第三方 Python 绑定,非常知名。
风险:GPL 协议,并不是和 Qt 一样的 LGPL。也就是说,使用 PyQt 开发的软件必须开源,除非购买商业授权。特别不建议使用。
没有看到 GitHub 仓库。
PySide Qt 官方 Python 绑定,也叫 Qt for Python。
据说当年 Qt 的持有者,Nokia 公司,找 Riverbank Computing 谈 PyQt 的授权问题,没有谈成,因而从新开发了这个项目。
没有 GitHub 仓库,代码可以在 官方 cgit 上看到。
注意:这里就指 pyside2,和更老的 pyside 区分开来。
PyGObject GTK 官方 Python 绑定
原来叫 PyGTK,多好,不知道为什么改成这个名字
https://gitlab.gnome.org/GNOME/pygobject
https://github.com/GNOME/pygobject
gooey 可以快速实现命令行 GUI 化。
logging 内部的服务级别:
DEBUG 10
INFO 20
WARNING 30
ERROR 40
CRITICAL 50
根据使用习惯,INFO 是重要信息,DEBUG 是普通信息。线上也是开到 DEBUG 级别。
然后调试信息也是通过 DEBUG 服务打印,然后通过 conf.DEBUG_MODE
来区分是不是要打印这种 DEBUG 级别的调试信息。
觉得不甚方便,想了一下,有两种思路:
方案一感觉相对合理一些,但是对于已有项目还是方案二好。
def trace(self, message, *args, **kwargs):
if self.isEnabledFor(TRACE):
self._log(TRACE, message, args, **kwargs)
TRACE = logging.TRACE = 5
logging.addLevelName(TRACE, 'TRACE')
logging.Logger.trace = trace
PyCrypto 是 Python 界最知名的加密模块,它提供了一系列的加密算法,包括对称加密、非对称加密、哈希算法、签名算法等。
不过有一个很大的问题:上一个版本 2.6.1 发布于 2013-10-18,已经很多年没有维护了。
PyCryptodome 是 PyCrypto 的分叉,该项目在统一套代码的基础上提供了两种包:pycryptodome
和 pycryptodomex
:
Crypto
名称下,后者丢掉了历史包袱,放弃对 PyCrypto 的兼容,所有代码都在 Cryptodome
名称下。
PyNaCl is a Python binding to libsodium, which is a fork of the Networking and Cryptography library.
有同事排查 Python 项目问题的时候指出一处 open
没有关闭可能会导致句柄泄露 Handle Leak。
PS: 句柄泄漏的危害:大量资源占用可能导致性能下降,甚至由于可打开文件数达到极限,服务无法继续向外提供服务。
我看了之后告诉他,此处函数退出之后句柄会自动关闭,他还不信,下去自己研究了一会儿,可能是百度一下,过一会儿说好像确实是这样,不过他仍然很疑惑,那么 with open
的作用是什么呢?
一般我们常用上下文管理的方式(with open
)来打开文件,这样可以自动关闭句柄,这是一个好的实践。
close
方法放在 finally
块中。进程退出时如果有没有关闭的句柄,
至少我看到 POSIX 中有相关规定,无论任何原因或任何方式的退出,都应该:
All of the file descriptors, directory streams, conversion descriptors, and message catalog descriptors open in the calling process shall be closed.
PS: 其中提到的:
后面三个是个啥?
import dataclasses
@dataclasses.dataclass
class User:
user_id: int
user_name: str
first_name: str
last_name: str
age: int = 0
def __post_init__(self):
self.full_name = f'{self.first_name} {self.last_name}'
u = User(1, 'admin', 'Jim', 'Green')
print(u)
# User(user_id=1, user_name='admin', first_name='Jim', last_name='Green', age=0)
print(u.full_name)
# Jim Green
u.full_name = 'Han Meimei'
print(u.full_name)
# Han Meimei
print(dataclasses.asdict(u))
# {'user_id': 1, 'user_name': 'admin', 'first_name': 'Jim', 'last_name': 'Green', 'age': 0}
print(dataclasses.astuple(u))
# (1, 'admin', 'Jim', 'Green', 0)
相当于:
class User:
def __init__(self, user_id: int, user_name: str, first_name: str, last_name: str, age: int = 0):
self.user_id = user_id
self.user_name = user_name
self.first_name = first_name
self.last_name = last_name
self.full_name = f'{self.first_name} {self.last_name}'
self.age = age
def __str__(self):
return f"{self.__class__.__name__}(user_id={self.user_id}, user_name='{self.user_name}', first_name='{self.first_name}', last_name='{self.last_name}, age={self.age}')"
def asdict(self):
return {
'user_id': self.user_id,
'user_name': self.user_name,
'first_name': self.first_name,
'last_name': self.last_name,
'age': self.age,
}
def astuple(self):
return self.user_id, self.user_name, self.first_name, self.last_name, self.age,
u = User(1, 'admin', 'Jim', 'Green')
print(u)
print(u.full_name)
u.full_name = 'Han Meimei'
print(u.full_name)
print(u.asdict())
print(u.astuple())
@dataclasses.dataclass(*, init=True, repr=True, eq=True, order=False,
unsafe_hash=False, frozen=False, match_args=True, kw_only=False, slots=False,
weakref_slot=False)
init
是否生成 __init__
方法(默认)repr
是否生成 __repr__
方法(默认)eq
是否生成 __eq__
方法(默认)order
是否生成 __lt__
,__le__
,__gt__
,__ge__
方法unsafe_hash
是否生成 __hash__
方法frozen
是否允许对字段赋值match_args
kw_only
是否仅关键词传参slots
是否生成 __slots__
方法weakref_slot
是否添加 __wrakref__
槽位(弱引用)