#59 Python DataClass

Python 2020-01-02

Python 3.7 新增了 dataclass，主要的好处是提供了一个类型定义模板，自动生成常见方法，减少一些常用的重复工作（__init__、__repr__、__eq__、__hash__ 和 __lt__ 等方法）。

示例

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__ 方法
    能 hash 的是不会变化对象，否则会出问题，需要谨慎使用。
  • frozen 是否允许对字段赋值
    可以使得数据类的实例变成不可变对象，类似于 Python 的 namedtuple。这意味着一旦实例化后，无法修改其中的属性值。
  • match_args
  • kw_only 是否仅关键词传参
  • slots 是否生成 __slots__ 方法
  • weakref_slot 是否添加 __wrakref__ 槽位（弱引用）
    指定 weakref_slot=True 而不同时指定 slots=True 将会导致错误。

dataclass 和 __slots__ 的区别

dataclass 作为代码模板，作用是简化类的定义
__slots__ 的作用是优化内存使用（dataclass 可以通过 slots=True 自动生成 __slots__ 定义）

from dataclasses import dataclass

# 使用 dataclass
@dataclass
class Person:
    name: str
    age: int
    email: str = None

# 创建数据类实例
p = Person(name="Alice", age=30)
print(p)  # Person(name='Alice', age=30, email=None)

# 使用 __slots__
class PersonWithSlots:
    __slots__ = ['name', 'age', 'email']

    def __init__(self, name, age, email=None):
        self.name = name
        self.age = age
        self.email = email

# 创建带 __slots__ 的实例
p2 = PersonWithSlots(name="Alice", age=30)
print(p2.name, p2.age)  # Alice 30

#58 Python sys.audit

Python 2019-11-06

在学习 subprocess 模块的时候看到代码中有一行：

sys.audit("subprocess.Popen", executable, args, cwd, env)

才了解到 Python 3.8 中新加了这样一个审计功能。

作用

提供一种机制，对 Python 程序执行过程中的部分关键操作进行干预，比如记录事件、引发异常、中止操作或完全终止进程。

部分关键操作发生的时候，Python 会调用我们定义的钩子方法。

注意：官方提示，恶意代码可以轻易绕过这些钩子，不要指望审计钩子能用来创建一个安全的 sandbox。
注意：sys.addaudithook 方法的调用会触发一个同名的事件。如果现有的钩子方法抛出 Runtime 异常，则不会添加这个钩子。

支持的审计事件

CPython 定义了一些事件：Audit events table，比如：

  • import
  • exec
  • compile
  • os.exec
  • os.listdir
  • os.rmdir
  • os.spawn
  • os.walk
  • os.system
  • subprocess.Popen

使用

sys.audit 方法是官方定义和执行的，Python 运行时负责抛出审计事件，我们程序只管使用系统方法 sys.addaudithook(hook) 来添加审计钩子处理这些事件。

import sys

def hook(event, args):
    print('%s: %r' % (event, args))

sys.addaudithook(hook)

import math
num = math.ceil(100 / 3)
print(num)

#56 Django 3.0 发布，开始支持异步功能

Python Django 2019-09-16

Django 项目组试图在保持向后兼容的基础之上，对阻塞部分进行改造，使之支持异步（通过装饰器的方式）。
包括 Session、Auth、ORM 与 Handlers 等。

ASGI 模式将 Django 作为原生异步应用程序运行，原有的 WSGI 模式将围绕每个 Django 调用运行单个事件循环，以使异步处理层与同步服务器兼容。

在这个改造的过程中，每个特性都会经历以下三个实现阶段：

  • Sync-only，只支持同步，也就是当前的情况
  • Sync-native，原生同步，同时带有异步封装器
  • Async-native，原生异步，同时带同步封装器

Django 3.0 开始提供运行 ASGI 应用支持，让 Django 逐渐具备异步功能。做了这一改动后，Django 现在会感知到异步事件循环，并将阻止从异步上下文调用标记为 “异步不安全” 的代码（例如 ORM 操作），如果开发者之前使用的是异步代码，则可能会触发。如果看到 SynchronousOnlyOperation 错误，可以仔细检查代码并将数据库操作移到同步子线程中。

其它方面，Django 现在支持 MariaDB 10.1 及更高版本；新的 ExclusionConstraint 类可以在 PostgreSQL 上添加排除约束；输出 BooleanField 的表达式现在可以直接在 QuerySet 过滤器中使用，而无需先注解然后对注解进行过滤；自定义枚举类型 TextChoices、IntegerChoices 和 Choices 现在可用作定义 Field.choices 的方法。

需要特别注意的是：自从 2.2 之后，Django 将不再支持 Python 3.5。

更新说明：https://docs.djangoproject.com/en/dev/releases/3.0

#52 Python DKIM 签名

Python Email DKIM 2019-04-16
  • pydkim
  • https://pypi.org/project/pydkim/
  • https://hewgill.com/pydkim/
  • NOTE: This page describes the last release of pydkim from 2008. The latest version is a fork found at dkimpy in Launchpad and is under active development.
  • 最新版本是 2008/06 发布的 v0.3
  • dkimpy
  • https://pypi.org/project/dkimpy/
  • https://launchpad.net/dkimpy
  • 最新版本是昨天发布的 v0.9.2
Date Version
2023-07-28 1.1.5
2023-05-12 1.1.4
2023-04-30 1.1.3
2023-04-09 1.1.2
2023-03-10 1.1.1
2023-02-25 1.1.0
2023-04-30 1.0.6
2020-08-09 1.0.5
2020-04-06 1.0.4
2020-01-15 1.0.3
2019-12-31 1.0.2
2019-12-15 1.0.1
2019-12-09 1.0.0
2019-12-24 0.9.6
2019-10-07 0.9.5
2019-09-25 0.9.4
2019-08-09 0.9.3
2019-04-15 0.9.2
2018-12-09 0.9.1
2018-10-30 0.9.0

说明

默认签名字段（28 个）：

cc, content-description, content-id, content-transfer-encoding, content-type,
date,
from,
in-reply-to,
list-archive, list-help, list-id, list-owner, list-post, list-subscribe, list-unsubscribe,
message-id, mime-version,
references, reply-to, resent-cc, resent-date, resent-from, resent-message-id, resent-sender, resent-to,
sender, subject,
to

d = dkim.DKIM('')
print(b', '.join(d.should_sign | d.frozen_sign))
b'list-unsubscribe, content-id, list-id, mime-version, resent-date, sender, cc, reply-to, content-type, list-owner, resent-message-id, resent-cc, resent-from, to, content-description, date, list-post, in-reply-to, content-transfer-encoding, from, references, list-help, subject, list-archive, resent-sender, list-subscribe, message-id, resent-to'
print(b', '.join(sorted(d.should_sign | d.frozen_sign)))
b'cc, content-description, content-id, content-transfer-encoding, content-type, date, from, in-reply-to, list-archive, list-help, list-id, list-owner, list-post, list-subscribe, list-unsubscribe, message-id, mime-version, references, reply-to, resent-cc, resent-date, resent-from, resent-message-id, resent-sender, resent-to, sender, subject, to'

示例

准备实验用的密钥对。

openssl genpkey -algorithm RSA -out /tmp/private_key.pem
openssl rsa -in /tmp/private_key.pem -check
openssl rsa -pubout -in /tmp/private_key.pem -out /tmp/public_key.pem

import dkim

domain = 'mail.markjour.com'
selector = 's20190416'

with open('/tmp/private_key.pem', 'rb') as f:
    privkey = f.read().strip()
# dkim.parse_pem_private_key(privkey)

message = """
Content-Transfer-Encoding: quoted-printable
Content-Type: text/html; charset=utf-8
Date: Mon, 24 Sep 2018 12:31:21 +0000 (UTC)
From: Admin <no-reply@mail.markjour.com>
Mime-Version: 1.0
Subject: Hello World
Message-ID: <n4F5zz24LXvYqPHVrZLPJokasT7MlLxYQx6g>
Reply-To: sender@mail.markjour.com
To: kwicoo@gmail.com
List-Unsubscribe: <mailto:unsubscribe@mail.markjour.com?p=Ahi2DRmdOnTdpsDzPClCPqbpwmFyjvGJV2xfJGWqw6eFEKRwI402QeoSsFrArTw1s48A59f60pLl0x71ojsQSWERnp3aMZA6YvEw>
X-SMTP-ID: c89cf6a5-22b7-4d1a-9bce-9f91a6be1bfb

HELLO WORLD
""".strip().encode()
# dkim.rfc822_parse(message)
# print(dkim.DKIM(message).default_sign_headers())
# [b'Content-Transfer-Encoding', b'Content-Type', b'Date', b'From', b'Mime-Version', b'Subject', b'Message-ID', b'Reply-To', b'To', b'List-Unsubscribe', b'From']

signature = dkim.sign(message, selector.encode(), domain.encode(), privkey)
print(signature.decode())
# DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=mail.markjour.com;
#  i=@mail.markjour.com; q=dns/txt; s=s20190416; t=1555392275;
#  h=content-transfer-encoding : content-type : date : from :
#  mime-version : subject : message-id : reply-to : to : list-unsubscribe
#  : from; bh=qg03cTlGc4OH4uPv7BGgoUyhgh23r+o1O6qzYOLixvA=;
#  b=sJ09G6hHPaP6AMp2mqUXjEZ+BfUFz0o6nbpXWxJ4/OG0o9ZwPSj8aJibZtJjTKP3k/TR/
#  6SD543V8iNw+JwwM+XLOUZa0iduK+QkedccqNl5Hcfc9UI/U11NoHz76B3csL9KE9tb40jF
#  mlLCuVUjci4HlOfEoKF8Ame8yWDHXVoNS/YT9/OSSc5q5q+qp6OX6PvzzxDomCHC6kbhOdv
#  Yc/KEXrMQ1JQ971pRUBNQK3eN7bV7g1BwXuMEuhdwDa4aZ4YYcakKywo4Oey7bIy1E7evZN
#  5rUitRExLH4dQNrhxoZd4c3QOjd4ROTwseAaMN10U/egzDXjcw2q0UUC1UKQ==

append_headers = [b'x-smtp-id']

d = dkim.DKIM(message)
include_headers = d.default_sign_headers()
include_headers.extend(append_headers)
signature = d.sign(selector.encode(), domain.encode(), privkey, include_headers=include_headers)
print(signature.decode())
# DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=mail.markjour.com;
#  i=@mail.markjour.com; q=dns/txt; s=s20190416; t=1555392275;
#  h=content-transfer-encoding : content-type : date : from :
#  mime-version : subject : message-id : reply-to : to : list-unsubscribe
#  : from : x-smtp-id; bh=qg03cTlGc4OH4uPv7BGgoUyhgh23r+o1O6qzYOLixvA=;
#  b=MTSeE8X3R+8bn+kkJaX5j/OKPMe+sdombmmwK5zME3SHBqiOLbxCwOGyh3qJKXdLpJlEg
#  pBnsDmNEjgC/rtBoclvnlCsaN7OFcZIe6ehfjwGeaw41r38Y8IgUQCkuN+IiL8FN1IiMI2f
#  kSayumwcOCAwmA4yJfu8n1v4W416jXt775YKR+1bt2Df1fNA6FnfoSMTqZl7rHn9zo76Efg
#  yvm7M0uT3uz0NZbJtqOnMFzRri9TEj4jYiCgsNaBYA9prbZlA02svoJx9qIJ2mKA+EcVpxK
#  IsEAY4ZXzXfhynKLeOYGK786ghiZrtsQYGbP6c1fAzTNy+fLJzRFozsV/wEQ==

headers = set(dkim.DKIM.SHOULD) | set(dkim.DKIM.FROZEN) | append_headers - set(dkim.DKIM.SHOULD_NOT)
signature = dkim.sign(message, selector.encode(), domain.encode(), privkey, include_headers=headers)
print(signature.decode())
# DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=mail.markjour.com;
#  i=@mail.markjour.com; q=dns/txt; s=s20190416; t=1555392275;
#  h=in-reply-to : x-smtp-id : content-type : cc : content-id : list-post
#  : date : resent-from : list-owner : list-id : to : content-description
#  : list-subscribe : message-id : sender : mime-version :
#  resent-message-id : list-help : content-transfer-encoding : resent-cc
#  : resent-date : list-unsubscribe : references : resent-sender : from :
#  list-archive : subject : resent-to : reply-to;
#  bh=qg03cTlGc4OH4uPv7BGgoUyhgh23r+o1O6qzYOLixvA=;
#  b=lLJafHJ8B/DoO4FncLp+BIHaPy4xsq7dRAWjzAvkRoDSwjcg3EloW0FsCXS45EkmwmBZC
#  Vks7zeOR1CS8oxcpauhxj1XnlwfcwLWtAQ3pogQTzNh4EEUFiNfgJTdXefAh7cpGHolQmy7
#  w2TBXDPx+Ikynw2tNnGOBduLWi+BH3Et8KGaskR4D9QHWSrk4pqeaNannNhDPUfE98d2fS3
#  kKBvqiEaTubQBdi8VXcl8J4R1SfdJZR2NfBJkPjJejlwTJaSytF2zyberpgflj0sEc8iHvM
#  2UQRcpxqm8GMRyzzKAXBSTzQhmaTOHntGokDTunNlUc/izFFRJm9SFiVq64g==

OpenDKIM

使用 opendkim-genkey 生成签名私钥和 DNS 配置文件：

$ opendkim-genkey --verbose --domain=mail.markjour.com --selector=s20190416 --directory=/tmp/
opendkim-genkey: generating private key
opendkim-genkey: private key written to s20190416.private
opendkim-genkey: extracting public key
opendkim-genkey: DNS TXT record written to s20190416.txt

域名配置好之后，可以使用 opendkim-testkey 检查：

$ opendkim-testkey -d test.markjour.com -k /tmp/private_key.pem -s s20190416 -v
opendkim-testkey: 's20190416._domainkey.test.markjour.com' record not found

#51 Python Supervisor 的简单配置

Python supervisord 2019-03-11 
name =
script-path-and-args =
execute-path =
python-path =

[program:<name>]
; process_name=<name>       ; 默认就是 program 名称

command = python <script-path-and-args>
directory = <execute-path>  ; 执行路径
environment=PYTHONPATH=<python-path>    ; 设置环境变量，逗号隔开
; user=

; killasgroup = false       ; 没用过
stopasgroup = true          ; 杀掉子进程，文档说是包含 killasgroup
; stopsignal=TERM           ; TERM, HUP, INT, QUIT, KILL, USR1, or USR2 中的一个
stopwaitsecs = 15

autostart = true
autorestart = true          ; true, false, unexpected
; exitcodes=0,2             ; 允许的退出码，否则会进入自动重启判断
; startretries = 3          ; 重启次数

numprocs=1
numprocs_start=0

loglevel = debug            ; critical, error, warn, info, debug, trace, blather
redirect_stderr = true
stdout_logfile = /var/log/<name>.log
stderr_logfile = /var/log/<name>.log

; stdout_logfile_maxbytes=1MB
; stdout_logfile_backups=10
; stdout_capture_maxbytes=1MB
; stdout_events_enabled=false

; stderr_logfile_maxbytes=1MB
; stderr_logfile_backups=10
; stderr_capture_maxbytes=1MB
; stderr_events_enabled=false

; 没用过
; umask=022
; priority=999
; serverurl=AUTO

另外，文档中看到一处有意思的用法，配置可以这样通过参数传递给进程：

[program:example]
command=/usr/bin/example --loglevel=%(ENV_LOGLEVEL)s
  1. 简单多进程就让 supervisor 控制
  2. 进程日志由进程自己控制，supervisor 只记录本身的运行与监控日志和进程漏出来的错误日志等。

#50 Python 解析 UserAgent

Python 浏览器 2019-02-08

GitHub 上搜索到的这些项目：

ua-parser 和 user-agents 这两个库数据漂亮些，所以，就选这两个库研究研究。

ua-parser

from ua_parser import user_agent_parser

ua_string = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2272.104 Safari/537.36'

parsed_string = user_agent_parser.Parse(ua_string)
pp.pprint(parsed_string)

user-agents