使用 Python 对数据进行压缩

之前在工作中遇到一个需求，需要在手机小程序端获取到微信小商店店铺的所有商品数据。由于当时我们没有在后台维护用户的商品数据，选择的解决方案是现场调用商品列表接口，然后缓存在 Redis 里。
鉴于 Redis 的内存还是比较宝贵的，而用户的商品数据（转化为 json 格式后）又是一些比较有规律的文本数据，比较适合进行数据压缩，于是我调研了一下 Python 中的数据压缩的方案。

zip标准库

Python 中有一个标准库zip,常用来处理 zip 格式文件，提供的函数和方法也更适合文件操作。在这个案例中我们的数据是通过 http 接口获取的，额外进行一些文件操作有些麻烦和多余，zip标准库并不适合这个场景。

zlib标准库

zlib是一个常用的压缩、解压库，使用了 deflate 算法。zip和unzip工具背后其实用的就是zlib。Python 内置了zlib标准库，提供的接口也很简洁易用（zlib.compress和zlib.decompress）。

1
2
3
4
5

import zlib
raw = b'abc' * 100
compressed = zlib.compress(raw)
print(f'compress ratio =  {len(compressed)/len(raw):.2}') # compress ratio =  0.05
assert zlib.decompress(compressed) == raw

前段时间遇到一个需要在本地开发环境定时执行某个脚本来同步数据的情景。

如果是线上业务的话，我一般会使用Celery Beat来执行定时任务。考虑到运行Celery Beat需要启动两个服务（celery worker和celery beat），还要管理虚拟环境并编写 Python 代码，用在简单的定时任务执行场景过于复杂。

说起定时任务，首先想到的就是crontab。于是我调研了一下crontab的使用方法，成功配置了定时任务，也在这里简单的介绍一下。

crontab 简介

crontab（守护进程为crond）是类 unix 平台下的一个定时任务管理工具，可用于周期性执行定时任务。

今天在公司处理了一个线上问题，涉及到在 Python 中处理语义化版本（Semantic Versioning），值得作为一个主题记录一下。

Node.js 的作者 Ryan Dahl 在 2018 年 JSConf EU 上发布了一篇演讲：10 Things I Regret About Node.js，讲述了他认为自己在设计 Node.js 时的十个（实际上演讲中只提到七个）失误，包括：没有坚持使用 Promise，使用了 GYP 构建系统，package.json 和 node_modules 的设计失误等。

在演讲中 Ryan Dahl 宣布了一个新项目：Deno，一个新的服务端 Javascript 运行时。经过两年多的发展，Deno 已经发布了 1.8 版本，也有了一个活跃的开发者社区。我们今天就来简单的了解一下 Deno。

Deno 简介

Deno is a simple, modern and secure runtime for JavaScript and TypeScript that uses V8 and is built in Rust.

Deno 使用 V8 引擎，由 Rust 构建，是一个简单、现代、安全的 Javascript 和 Typescript 运行时。

Deno 的主要特性有：

• 默认安全。外部代码没有文件系统、网络、环境的访问权限，除非显式开启。

• 支持开箱即用的 TypeScript 的环境。

• 只分发一个独立的可执行文件 (deno)。

• 有着内建的工具箱，比如一个依赖信息查看器 (deno info) 和一个代码格式化工具 (deno fmt)。

• 有一组经过审计的标准模块，保证能在 Deno 上工作。

• 脚本代码能被打包为一个单独的 JavaScript 文件。

Python 3.10 版本还在开发之中，目前释出的 dev 版本实现了新语法特性 Structural Pattern Matching(PEP 634)：可以利用match语句和case语句匹配对象的不同模式，并应用不同的行为。

我先前自己尝试体验了一下Structural Pattern Matching语法（使用pyenv安装dev版本 Python 3.10），感觉很好用的，并且有很大的发挥空间。

Structural pattern matching has been added in the form of a match statement and case statements of patterns with associated actions. Patterns consist of sequences, mappings, primitive data types as well as class instances. Pattern matching enables programs to extract information from complex data types, branch on the structure of data, and apply specific actions based on different forms of data.

官方介绍

语法

1
2
3
4
5
6
7
8
9

match subject:
    case <pattern_1>:
        <action_1>
    case <pattern_2>:
        <action_2>
    case <pattern_3>:
        <action_3>
    case _:
        <action_wildcard>

最简单的模式匹配

我们可以使用match case直接匹配字面量。

1
2
3
4
5
6
7
8
9
10

def http_error(status):
    match status:
        case 400:
            return "Bad request"
        case 404:
            return "Not found"
        case 418:
            return "I'm a teapot"
        case _:
            return "Something's wrong with the Internet"

最近使用 FastAPI 框架开发了一个 WEB 服务。

为了充分利用 FastAPI 作为一个 ASGI 框架的原生异步支持特性，很多业务代码也改成了异步函数，并且使用了异步的 HTTP 库httpx和 MongoDB 的异步 Python drivermotor。

由此带来的一个问题就是异步 Python 代码的单元测试的编写问题。

测试异步函数

编写测试代码

Python 的异步函数返回的是一个协程对象（coroutine），需要在前面加await才能获取异步函数的返回值，而只有在异步函数中才能使用await语句，这也意味着一般异步函数的测试代码本身也需要是一个异步函数。

1
2
3
4
5
6

async def add(a:int, b:int):
    return a + b

async def testAdd():
    ret = await add(1, 2)
    assert ret == 3

运行测试代码

与 Javascript 不同，Python 的异步代码需要显示地运行在事件循环中。

Python3.7 以上的版本中可以直接调用asyncio.run函。

如果使用的是更早的 Python 版本，就需要指定一个事件循环对象来运行异步代码。

1
2
3
4
5
6
7
8

import asyncio

# Python3.7+
asyncio.run(testAdd())

# Python3.6
loop = asyncio.new_event_loop()
loop.run_until_complete(testAdd())

最近创建了几个 Python 的开源项目（u-coreutils、auto-logger、fastapi-redis-session）。

我在这些项目中开始使用poetry工具进行依赖管理，体验还是相当不错的。

poetry 简介

Poetry是一个基于pyproject.toml的工具，利用poetry可以更加方便地进行 Python 项目的开发、构建、发布和依赖管理。

接下来通过一个例子来简单介绍一下 poetry 的使用。

初始化项目

我们将创建一个库greet，提供一个greet函数来输出”Hello, World”，代码和文件结构都很简单。

1
2
3

# project/greet/__init__.py
def greet(name:str='World'):
    print(f'Hello, {name}')

在project目录下运行poetry init(可以通过pip install -U poetry进行poetry工具的安装)，会有很多提示信息用于确认项目的名称、版本、作者、协议、Python 版本，也可以在init过程中指定项目的依赖和dev依赖。

poetry init运行成功后，solution目录下会生成一个pyproject.toml文件。

可以手动设置一些项目的其他信息，例如仓库地址或者README文件的位置。

对数据进行排序是一个很常见的需求，但有时候我们并不需要对完整的数据进行排序，只需要排前几的数据，也就是经典的 Top-K 问题。

Top-K 问题的经典解法有两种：一种是脱胎于快速排序（Quick Sort）的快速选择（Quick Select）算法，核心思路是在每一次Partion操作后下一次递归只操作前K项数据。另一种是基于堆排序的方法。

Python 中有两个标准库可以原生的支持堆排序(优先队列)，分别是heapq和PriorityQueue(queue)。

heapq

heapq标准库提供了一些工具函数用来对list对象实现二叉堆的各种操作（就地修改list对象）。简单的用法如下：

建堆

1
2
3
4
5
6

import heapq

# 可以用过random.shuffle函数创造乱序数组
arr = [4, 0, 3, 1, 6, 5, 9, 7, 8, 2]
heapq.heapify(arr)
assert arr == [0, 1, 3, 4, 2, 5, 9, 7, 8, 6]

获取堆顶元素

1
2

assert heapq.heappop(arr) == 0
assert arr == [1, 2, 3, 4, 6, 5, 9, 7, 8]

背景

最近有一个业务需要同时写两个表，并且需要保证数据的正确性。我们的项目线上的 MongoDB 版本是 4.0，查了一下发现 4.0 版本的 MongoDB 已经支持副本集中的事务了，于是做了一下调研。

准备环境

鉴于 MongoDB 的事务支持需要在副本集中才生效，我用 Docker-Compose 部署了一个一主两从结构的 4.0 版本 MongoDB 副本集(部署过程参考了文章)。

初始化数据

假设有两个表，订单表(orders)和库存表(inventory), 对于一个下单行为：创建一个购买了商品 A x 件的订单，需要增加一条 orders 表记录表明购买了 x 件商品 A，同时商品 A 对应的库存表记录的库存也要减少 x 件。

1
2
3
4
5
6

import pymongo

client = pymongo.MongoClient(host="mongodb://root:123@localhost:37011")

db = client.get_database("test")
db.inventory.insert_one({"sku": "abc", "qty": 100000})

在代码中我们初始化了十万件商品 abc。

使用装饰器、魔法方法和元类机制自动记录 Python 函数和方法调用日志

在软件开发中，日志记录一直是一件让程序员很矛盾的事情。在编写代码的时候，不能确定需要将哪些信息记录日志；真正出了问题或者想统计分析一些用户行为和系统状态的时候，又发现没有充足的日志。
一个不错的解决方案是利用自动化机制记录关键函数和方法的调用记录。今天我们来看几种自动记录 Python 函数和方法调用日志的实现手段。

手动记录日志

这是最标准和常见的方法，在需要的时候手动记录一些信息，优点是完全可控，缺点是容易缺失一些真正重要的信息的记录。

1
2
3
4
5
6
7

import logging
logger = logging.getLogger(__name__)

def func(*args, **kwargs):
    logger.info(f'Call func with {args, kwargs}')

func(1, 2, 3, a=True, b='arg')

1

INFO:__main__:Call func with ((1, 2, 3), {'a': True, 'b': 'arg'})

使用装饰器记录函数和方法调用日志

我们可以利用 Python 的装饰器记录函数和方法调用的参数和返回值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

def func_logger(func):

    def inner(*args, **kwargs):
        ret = func(*args, **kwargs)
        logger.info(f'Call func {func.__name__} with {args, kwargs} returns {ret}')
        return ret

    return inner


@func_logger
def add(a, b):
    return a + b

add(1,2)
add(1, b=2)
add(a=1, b=2)

1
2
3

INFO:__main__:Call func add with ((1, 2), {}) returns 3
INFO:__main__:Call func add with ((1,), {'b': 2}) returns 3
INFO:__main__:Call func add with ((), {'a': 1, 'b': 2}) returns 3