Python 3 进阶速查

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 模块级变量，与 make_counter() 内部的 count 是两个不同的绑定。
count = 100


def make_counter(start=0):
    count = start

    def increment():
        # 更新外层 make_counter() 的 count，而不是模块级的 count。
        nonlocal count
        count += 1
        return count

    return increment


counter = make_counter(10)
print(counter())
print(counter())
print(count)

# 输出：
# 11
# 12
# 100

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

functions = [lambda: i for i in range(3)]
print([function() for function in functions])

# 默认参数在创建函数时求值，用它保存每轮的整数值。
functions = [lambda i=i: i for i in range(3)]
print([function() for function in functions])

# 输出：
# [2, 2, 2]
# [0, 1, 2]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from functools import wraps


def trace(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        print(f"调用 {function.__name__}")
        return function(*args, **kwargs)

    return wrapper


@trace
def add(left, right=0):
    """返回两个数的和。"""
    return left + right


print(add(2, right=3))
print(add.__name__)
print(add.__doc__)

# 输出：
# 调用 add
# 5
# add
# 返回两个数的和。

1
2
3
4

# 调用 add
# 5
# wrapper
# None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from functools import wraps


def label(name):
    def decorate(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            print(f"进入{name}")
            result = function(*args, **kwargs)
            print(f"离开{name}")
            return result

        return wrapper

    return decorate


@label("外层")
@label("内层")
def greet():
    return "你好"


print(greet())

# 输出：
# 进入外层
# 进入内层
# 离开内层
# 离开外层
# 你好

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from functools import update_wrapper


class CountCalls:
    def __init__(self, function):
        update_wrapper(self, function)
        self.function = function
        self.calls = 0

    def __call__(self, *args, **kwargs):
        self.calls += 1
        return self.function(*args, **kwargs)


@CountCalls
def add(left, right=0):
    return left + right


print(add(2, right=3))
print(add(4, 5))
print(add.calls)
print(add.__name__)

# 输出：
# 5
# 9
# 2
# add

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class Stock:
    def __init__(self, quantity):
        self.quantity = quantity

    @property
    def quantity(self):
        return self._quantity

    @quantity.setter
    def quantity(self, value):
        if value < 0:
            raise ValueError("库存不能为负数")
        self._quantity = value


stock = Stock(5)
stock.quantity = 3
print(stock.quantity)

try:
    stock.quantity = -1
except ValueError as error:
    print(f"ValueError: {error}")

# 输出：
# 3
# ValueError: 库存不能为负数

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

class Stock:
    def __init__(self, quantity):
        self.quantity = quantity

    def get_quantity(self):
        return self._quantity

    def set_quantity(self, value):
        if value < 0:
            raise ValueError("库存不能为负数")
        self._quantity = value

    quantity = property(get_quantity, set_quantity)


stock = Stock(5)
stock.quantity = 3
print(stock.quantity)

# 输出：3

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from io import StringIO


class TextReader:
    def __init__(self, text):
        self.text = text

    def __enter__(self):
        self.reader = StringIO(self.text)
        return self.reader

    def __exit__(self, exc_type, exc_value, traceback):
        self.reader.close()
        return False


try:
    with TextReader("hello") as reader:
        print(reader.read())
        raise ValueError("读取中断")
except ValueError as error:
    print(f"ValueError: {error}")

print(reader.closed)

# 输出：
# hello
# ValueError: 读取中断
# True

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from contextlib import contextmanager
from io import StringIO


@contextmanager
def text_reader(text):
    # StringIO 是内存中的文本流，本例不创建磁盘文件。
    reader = StringIO(text)
    try:
        yield reader
    finally:
        reader.close()


with text_reader("hello") as reader:
    print(reader.read())

print(reader.closed)

# 输出：
# hello
# True

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

def flatten(groups):
    for group in groups:
        # 依次产出当前 group 的元素；空列表没有元素可产出。
        yield from group


numbers = flatten([[1, 2], [3], []])
# 只取第一个数字，生成器暂停在这里。
print(next(numbers))
# 从暂停处继续，将剩余数字收集成列表。
print(list(numbers))
# 已经读完，不会自动回到开头。
print(list(numbers))

# 输出：
# 1
# [2, 3]
# []

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

from io import StringIO


def nonempty_lines(lines):
    for line in lines:
        text = line.strip()
        if text:
            yield text


with StringIO("第一行\n\n第二行\n") as source:
    for text in nonempty_lines(source):
        print(text)
        break

print(source.closed)

# 输出：
# 第一行
# True

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import copy

original = [[1, 2], [3]]
alias = original
shallow = copy.copy(original)
deep = copy.deepcopy(original)

# shallow[0] 与 original[0] 是同一个子列表，两边都能看到新增的 9。
shallow[0].append(9)
# 只给 shallow 的外层列表加一项，不会给 original 的外层加项。
shallow.append([4])

print(alias is original)
print(shallow is original)
print(shallow[0] is original[0])
print(deep[0] is original[0])
print(original)
print(shallow)
print(deep)

# 输出：
# True
# False
# True
# False
# [[1, 2, 9], [3]]
# [[1, 2, 9], [3], [4]]
# [[1, 2], [3]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from copy import replace
from dataclasses import FrozenInstanceError, dataclass


@dataclass(frozen=True)
class Config:
    host: str
    port: int


original = Config("localhost", 8000)
try:
    # frozen=True 禁止修改这个已有实例的字段。
    original.port = 9000
except FrozenInstanceError as error:
    print(f"FrozenInstanceError: {error}")

# 创建修改了 port 的新实例，原实例保持不变。
updated = replace(original, port=9000)
print(original)
print(updated)

# 输出：
# FrozenInstanceError: cannot assign to field 'port'
# Config(host='localhost', port=8000)
# Config(host='localhost', port=9000)

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

from dataclasses import dataclass


@dataclass
class Config1:
    host: str
    port: int


# 实例化新定义的 Config1，不要误用前面冻结的 Config。
original1 = Config1("localhost", 8000)
original1.port = 9000
print(original1)

# 输出：Config1(host='localhost', port=9000)

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

import re

# 自定义账号规则：小写字母开头，其后允许小写字母、ASCII 数字和下划线；总长 3～16。
account_pattern = re.compile(r"[a-z][a-z0-9_]{2,15}")
print(account_pattern.fullmatch("user_01") is not None)
print(account_pattern.fullmatch("user_01!") is not None)

match = re.search(r"order=(?P<number>[0-9]+)", "user=ada order=2048")
if match:
    print(match.group(0))
    print(match.group("number"))
    print(type(match.group("number")).__name__)

# 输出：
# True
# False
# order=2048
# 2048
# str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import re

print(re.fullmatch(r"\d+", "１２３") is not None)
print(re.fullmatch(r"[0-9]+", "１２３") is not None)
print(re.sub(r"\s+", " ", "hello\t  world"))

# 输出：
# True
# False
# hello world

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

import re

print(re.fullmatch(r"([a-z]+)-\1", "python-python") is not None)
print(re.fullmatch(r"([a-z]+)-\1", "python-java") is not None)
print(re.fullmatch(r"(?P<word>[a-z]+)-(?P=word)", "python-python") is not None)

# 重复写两次相同的字符规则，并不要求两次匹配到的文本相同。
print(re.fullmatch(r"([a-z]+)-([a-z]+)", "python-java") is not None)

# 输出：
# True
# False
# True
# True

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from threading import Lock, Thread

count = 0
lock = Lock()


def increment(times):
    global count
    for _ in range(times):
        # 两个线程必须使用同一把锁，保护完整的“读取、加一、写回”。
        with lock:
            count += 1


threads = [Thread(target=increment, args=(1000,)) for _ in range(2)]
for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

print(count)

# 输出：2000

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

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import get_context


def square(number):
    return number ** 2


if __name__ == "__main__":
    # 显式使用 spawn：启动新解释器，让示例跨平台行为更容易理解。
    context = get_context("spawn")
    with ProcessPoolExecutor(max_workers=2, mp_context=context) as executor:
        results = list(executor.map(square, [2, 3, 4]))
    print(results)

# 输出：[4, 9, 16]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio


async def fetch_label(name):
    # 仅模拟一次异步等待，不访问网络，也不是性能测试。
    await asyncio.sleep(0.01)
    return f"完成 {name}"


async def main():
    async with asyncio.TaskGroup() as group:
        first = group.create_task(fetch_label("A"))
        second = group.create_task(fetch_label("B"))

    # 离开任务组后两项都已完成；按固定顺序读取结果。
    print(first.result())
    print(second.result())


if __name__ == "__main__":
    asyncio.run(main())

# 输出：
# 完成 A
# 完成 B

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    # AF_INET 使用 IPv4；127.0.0.1 仅本机可访问，端口 0 由系统分配。
    server.bind(("127.0.0.1", 0))
    server.listen(1)
    server.settimeout(2)

    with socket.create_connection(server.getsockname(), timeout=2) as client:
        connection, _ = server.accept()
        with connection:
            connection.settimeout(2)
            client.sendall("你好".encode("utf-8"))
            client.shutdown(socket.SHUT_WR)

            chunks = []
            # := 将本次读取结果绑定到 chunk，同时检查是否到达流结束。
            while chunk := connection.recv(2):
                chunks.append(chunk)

print(b"".join(chunks).decode("utf-8"))

# 输出：你好

1
2
3
4
5

text = "hello"
if (length := len(text)) > 3:
    print(length)

# 输出：5

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from urllib.parse import parse_qs, urlsplit

url = urlsplit("https://example.com/search?q=python&page=2#intro")
print(url.scheme, url.hostname)
print(url.path)
print(parse_qs(url.query))
print(url.fragment)

# 输出：
# https example.com
# /search
# {'q': ['python'], 'page': ['2']}
# intro

1
2
3

GET /search?q=python HTTP/1.1
Host: example.com
Accept: application/json