最近创建了几个 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

使用 docker-compose 部署 flask 应用

为了让开发环境、测试环境和线上环境统一，使用 docker 镜像部署应用是一个不错的方案。裸用 docker-cli 管理服务过于麻烦，在开发机上 k8s 又凭空增加了很多复杂性，使用 docker-compose 部署应用是一个不错的折中方案。

Docker-compose 是 docker 官方推出的一个管理多容器服务的工具。通过 Compose，您可以使用 YML 文件来配置应用程序需要的所有服务。然后，使用一个命令，就可以从 YML 文件配置中创建并启动所有服务。其主要特性有：

1. 使用一个 Host 启动多个服务
2. 容器启动时持久化数据卷到宿主机
3. 只在镜像发生变化时重新创建容器
4. 更方便地管理环境变量

何为”更好”的代码本身就是仁者见仁，我们在这里主观地选择一个评价标准：

代码要满足安全性 ,可用性 ,可维护性 ,简洁性 , 高性能的要求，这几项的重要性递减。

毫无疑问编写代码本身是一件很难的事，但是在遵循一些指导原则的话，我们可以相对编写出更好的代码。

背景

开发人员每天运行最多的命令是哪一个？npm,mvn,make…或者是大家都猜到的————git。

我们在完成一个需求的开发，或者修复一个问题后，往往会进行如下一系列git命令：

1
2
3

git add .
git commit -m "Feature A (OR Fix BUG B)"
git push

另外还有一些命令如git status,git fetch,git log,git checkout也是我们经常要用到的。

其实大家完全可以把一些常用的git命令设置成git别名，方便日常使用。

初识 Python3.7 的 dataclasses 标准库

最近在进行一个新的后端项目时想初步应用一下领域驱动设计的思想。DDD 开发需要对一个领域对象进行各种操作，而不是把业务数据包在dict里在 action 层,repo 层中传来传去。如何方便高效地定义实体类成为一个重要前提。Python 3.7 版本引入的新标准库 dataclasses 可以帮助我们解决这个问题。

常用国内镜像地址和使用方式

Pypi

镜像地址

使用方式

临时使用

1

pip install flask -i https://pypi.doubanio.com/simple/

比较 Python 中的列表推导式和 map(),reduce()函数

对一个列表（迭代器）中的元素进行批量处理是一个很常见的业务需求，在 Python 中，一般有三种解决方案：for循环，列表推导式，或者map(),filter()函数。

例如我们计算一下 100 以内奇数的平方和。

1
2
3
4
5

# for loop
total = 0
for x in range(100):
    if x % 2:
        total += x * x

1
2

# list comprehension
sum(x * x for x in range(100) if x % 2)

1
2

# map(), filter()
sum(map(lambda x: x * x, filter(lambda x: x % 2, range(100))))