1.python 基本命令
python环境在conda中,使用anaconda prompt
1.使用conda创建python环境,安装python版本conda create -n 环境名 python=3.8 # 创建python3.8环境-n: 指定环境名称python=3.8: 指定python版本
2.查看当前已有的环境conda env list # 列出所有环境
3.激活环境conda activate py38 # 进入python3.8环境
4.安装依赖包conda install 包名 # 安装包conda install -r requirements.txt # 从文件安装依赖包
5.查看当前已有的环境conda env list # 列出所有环境
6.退出环境conda deactivate # 退出当前环境
进入环境后,输入以下命令:常用的python命令:
python -version # 查看python版本python -m pip list # 查看已安装的包python -m pip install 包名 # 安装包python -m pip install -r requirements.txt # 从文件安装依赖包python -m pip freeze > requirements.txt # 将当前环境的包列表输出到文件python -m pip uninstall 包名 # 卸载包python -m pip show 包名 # 显示包信息
python # 进入交互式命令行exit() # 退出交互式命令行python 文件名.py # 运行python文件2. 基础语法
2.1 注释与缩进
python使用#进行单行注释,使用''' '''或""" """进行多行注释。 python使用缩进来表示代码块,通常使用4个空格(也可以使用tab,但不推荐)表示一个缩进级别。缩进必须一致,否则会导致IndentationError错误。 函数之间或类的方法之间用空行分隔,表示一段新的代码的开始。
分支语句中if True: print ("Answer") print ("True")else: print ("Answer") print ("False") # 缩进不一致,会导致运行错误
再比如循环中for i in range(5): print(i) # 这是循环体,必须缩进 print("循环内的代码") # 这是循环体,必须缩进print("循环结束") # 这是循环外的代码,不需要缩进
多行语句Python 通常是一行写完一条语句,但如果语句很长,我们可以使用反斜杠 \ 来实现多行语句,total = item_one + \ item_two + \ item_three2.2 变量与数据类型
python是动态类型语言,变量不需要声明类型,直接赋值即可。 常见的数据类型有:整数(int)、浮点数(float)、字符串(str)、布尔值(bool)、列表(list)、元组(tuple)、字典(dict)等。
# 变量赋值x = 10 # 整数y = 3.14 # 浮点数name = "Alice" # 字符串is_valid = True # 布尔值# 数据类型示例my_list = [1, 2, 3, 4, 5] # 列表my_tuple = (1, 2, 3) # 元组my_dict = {"name": "Alice", "age": 30} # 字典# 输出变量类型print(type(my_tuple)) # <class 'tuple'>为什么 Python 不需要声明变量类型?
因为 Python 是动态类型语言,变量类型在运行时根据赋值自动确定,无需预先声明。
变量本身没有固定类型,类型属于它所引用的对象。
动态类型机制如何实现?
Python 主要通过对象引用和运行时类型信息实现动态类型:
- 变量是引用:执行
a = 42时,Python 创建整数对象并让a指向它;执行a = "Hello"时,a改为指向字符串对象。 - 类型属于对象:对象内部保存类型信息。CPython 中,
PyObject的ob_type指针指向具体类型,如PyLong_Type或PyUnicode_Type。 - 运行时动态派发:执行
a + b时,解释器运行时检查对象类型,并查找相应的运算实现(如nb_add或__add__)。
注意:动态类型不等于没有类型,而是类型检查推迟到运行时。
详细介绍部分数据类型
2.2.1字符串
与c++中的字符串类似,python中的字符串是不可变的,可以使用单引号、双引号或三引号来定义字符串。
str1 = "Hello, World!" # 双引号定义字符串常用的字符串操作方法有:str1.upper() # 转换为大写str1.lower() # 转换为小写str1.strip() # 去除首尾空格str1.split(",") # 按照逗号分割字符串str1.replace("World", "Python") # 替换字符串中的内容str1.find("World") # 查找子字符串的位置,返回索引,如果找不到返回-1str1.isalpha() # 判断字符串是否只包含字母,返回True或Falsestr1[0] # 获取字符串的第一个字符
#字符串切片str1[start:end:step] #切片操作,获取字符串的子串,start表示起始索引,end表示结束索引(左闭右开),step表示步长,默认值为1eg: str1[0:5] # 获取字符串的前5个字符,结果为 "Hello" str1[7:] # 获取字符串从索引7开始到末尾的子串,结果为 "World!" str1[-1] # 获取字符串的最后一个字符 str1[-5:] # 获取字符串的最后5个字符,结果为 "orld!" str1[0:-1] # 获取从开始到倒数第二个字符的子串,结果为 "Hello, World" str[1:5:2] # 获取字符串从索引1到索引5(不包含)的字符,步长为2,结果为 "el" str1[::2] # 获取字符串的偶数索引字符,结果为 "Hlo ol!" str1[::-1] # 获取字符串的反转,结果为 "!dlroW ,olleH"
#字符串输出print(str1 * 2) # 输出字符串两次print(str1 + '你好') # 连接字符串
print('hello\nrunoob') # 使用反斜杠(\)+n转义特殊字符输出:hellorunoob
print(r'hello\nrunoob') # 在字符串前面添加一个 r,表示原始字符串,不会发生转义输出:hello\nrunoob
# print默认换行输出,不换行输出需要在变量末尾加上 end="":print( str1, end=" " )print("这是下一条内容")输出:Hello, World! 这是下一条内容2.2.2 列表
列表是python中最常用的数据类型之一,是一个有序的可变集合,可以包含不同类型的元素。列表使用方括号 [] 定义,元素之间用逗号分隔。
与c++与java中的数组不同,python中的列表可以动态调整大小,可以包含不同类型的元素,并且提供了丰富的方法来操作列表。类似于java中的ArrayList和c++中的vector,但更为灵活和强大。
# 创建列表my_list = ['abcd', 786, 2.23, 'runoob', 70.2] # 避免使用 list 作为变量名,会覆盖内置类型tinylist = [123, 'runoob']
print(my_list) # 打印整个列表:['abcd', 786, 2.23, 'runoob', 70.2]print(my_list[0]) # 打印第一个元素(索引 0):abcdprint(my_list[1:3]) # 打印索引 1 和 2 的元素(不含索引 3):[786, 2.23]print(my_list[2:]) # 打印从索引 2 开始到末尾的所有元素:[2.23, 'runoob', 70.2]print(tinylist * 2) # 重复打印 tinylist 两次:[123, 'runoob', 123, 'runoob']print(my_list + tinylist) # 拼接两个列表
#list常用操作
#1.访问列表元素my_list[0] # 通过索引访问列表元素,索引从0开始,负数索引表示从末尾开始计数
#2.修改列表元素my_list[0] = "new_value" # 通过索引修改列表元素print(my_list) # 打印修改后的列表 [new_value, 786, 2.23, 'runoob', 70.2]
#3.添加元素my_list.append("new_element") # 在列表末尾添加新元素
#4.删除元素my_list.remove(786) # 删除列表中第一个匹配的元素my_list.pop(0) # 删除指定索引的元素,并返回该元素,默认删除最后一个元素
#5.列表长度my_list_length = len(my_list) # 获取列表的长度(元素个数) #len()是python内置函数,返回对象的长度或元素个数
#6.列表切片#my_list[start:end:step] # 切片操作,获取列表的子列表,start表示起始索引,end表示结束索引(左闭右开),step表示步长,默认值为1
#7.列表排序#my_list.sort() # 对列表进行升序排序#warning 注意列表中的元素必须是可比较的类型,否则会抛出 TypeError 异常
#my_list.sort(reverse=True) # 对列表进行降序排序#my_list_sorted = sorted(my_list) # 返回一个新的升序排序的列表,不改变原列表 #sorted()是python内置函数,返回一个新的排序后的列表
#8.列表反转my_list.reverse() # 反转列表中的元素顺序my_list_reversed =reversed(my_list) # 返回一个反转后的迭代器,不改变原列表 #reversed()是python内置函数,返回一个反转后的迭代器print(str(list(my_list_reversed))) # 将迭代器转换为列表并打印
#9.元素计数count = my_list.count('runoob') # 统计列表中某个元素的出现次数2.2.3 元组
元组(tuple)与列表类似,但元组是不可变的,一旦创建就不能修改。元组使用圆括号 () 定义,元素之间用逗号分隔。元组可以包含不同类型的元素。 如果与c++中的数组相比,元组更像是一个固定大小的数组,不能动态调整大小,但可以包含不同类型的元素。 那为什么有了列表还要有元组呢? 主要是因为元组的不可变性使得它在某些场景下更安全,比如作为字典的键,或者在多线程环境中使用时,元组可以避免数据被意外修改。
my_tuple = ('abcd', 786, 2.23, 'runoob', 70.2) # 避免使用 tuple 作为变量名tinytuple = (123, 'runoob')
print(my_tuple) # 输出完整元组print(my_tuple[0]) # 输出第一个元素:abcdprint(my_tuple[1:3]) # 输出索引 1 和 2 的元素:(786, 2.23)print(my_tuple[2:]) # 输出从索引 2 开始的所有元素print(tinytuple * 2) # 输出两次 tinytupleprint(my_tuple + tinytuple) # 连接两个元组
#warning 元组虽然是不可变的,但如果元组中包含可变对象(如列表),则这些可变对象仍然可以被修改。my_tuple_with_list = (1, 2, [3, 4])my_tuple_with_list[2][0] = 99 # 修改元组中的列表元素
#同时元组支持切片操作,类似于列表,但元组的切片操作会返回一个新的元组,而不是列表。#元组没有append()、remove()等方法,因为它们是不可变的,但可以使用 + 运算符连接元组,或者使用 * 运算符重复元组。使用元组的场景
- 不可变性:当你需要一个不可变的序列时,使用元组可以确保数据不会被意外修改。
- 哈希性:元组可以作为字典的键,而列表不行,因为元组是不可变的。
- 性能:元组通常比列表更节省内存和更快,因为它们是不可变的,Python可以对它们进行优化。 元组通常。比列表更节省内存和更快,因为它们是不可变的,Python可以对它们进行优化。 创建速度更快:创建元组比创建列表快。 内存占用更小:元组占用的内存空间比同等元素的列表少。 缓存机制:Python 内部会缓存一些常用的空元组或小元组,避免重复分配内存
2.2.4 集合
Python 中的集合(Set)是一种无序、可变的数据类型,用于存储唯一的元素。 集合中的元素是唯一的,不能重复。 集合使用大括号 {} 定义,元素之间用逗号分隔。集合支持数学上的集合操作,如并集、交集、差集等。 类似于c++中的std::set和java中的HashSet,但python的集合更为灵活和强大。
sites = {'Google', 'Taobao', 'Runoob', 'Facebook', 'Zhihu', 'Baidu'} #使用的是字面量语法(大括号 {})。这是直接定义一个包含多个字符串元素的集合。
print(sites) # 输出集合(无序,重复元素会被自动去掉)
# 成员测试if 'Runoob' in sites: print('Runoob 在集合中')else: print('Runoob 不在集合中')
# set 可以进行集合运算a = set('abracadabra') #set() 接收一个字符串时,它会将字符串打散,把里面的每一个字符当作独立的元素放入集合中。b = set('alacazam')
print(a) # a 中的唯一字符
print(a - b) # a 和 b 的差集(在 a 中但不在 b 中)print(a | b) # a 和 b 的并集(在 a 或 b 中)print(a & b) # a 和 b 的交集(同时在 a 和 b 中)print(a ^ b) # a 和 b 的对称差集(在 a 或 b 中,但不同时存在)
#常用操作1.添加元素sites.add('Twitter') # 添加元素
2.删除元素sites.remove('Baidu') # 删除元素,如果元素不存在会抛出 KeyError 异常
3.清空集合sites.clear() # 清空集合
4.集合长度len(sites) # 获取集合的长度(元素个数)
5.集合遍历for site in sites: print(site) # 遍历集合中的元素
6.元素是否存在if 'Google' in sites: print('Google 在集合中')else: print('Google 不在集合中')使用集合的场景:
- 去重:当你需要从一个列表或其他可迭代对象中去除重复元素时,集合是一个很好的选择。
- 数学运算:如果你需要进行集合的交集、并集、差集等操作,集合提供了方便的方法。
- 快速查找:集合的成员测试(in 操作)通常比列表更快,尤其是在处理大量数据时。
2.2.5 字典
字典是一种映射类型,用 {} 标识,它是一个 键(key) : 值(value) 的集合。键(key) 必须使用不可变类型,且在同一个字典中键必须是唯一的。 注意: 1.Python 3.7 起,字典会保持元素的插入顺序,不再是无序的。如果需要有序字典的特性,直接使用普通 dict 即可。
2.字典的键与值可以包含任意类型的数据。
my_dict = {}my_dict['one'] = "1 - 菜鸟教程"my_dict[2] = "2 - 菜鸟工具"
tinydict = {'name': 'runoob', 'code': 1, 'site': 'www.runoob.com'}
print(my_dict['one']) # 输出键为 'one' 的值print(my_dict[2]) # 输出键为 2 的值print(tinydict) # 输出完整的字典print(tinydict.keys()) # 输出所有键print(tinydict.values()) # 输出所有值
#元组作为字典的键my_dict_with_tuple = {('a', 'b'): 1, ('c', 'd'): 2} # 使用元组作为字典的键print(my_dict_with_tuple[('a', 'b')]) # 输出键为 ('a', 'b') 的值:1
#元组常用操作1.访问元素my_dict_with_tuple[('a', 'b')] = 3 # 修改键为 ('a', 'b') 的值为 3
2.删除元素del my_dict_with_tuple[('c', 'd')] # 删除键为 ('c', 'd') 的键值对
3.遍历字典for key, value in my_dict_with_tuple.items(): print(f"键: {key}, 值: {value}") # 遍历字典中的键值对
4.字典长度len(my_dict_with_tuple) # 获取字典的长度(键值对的数量)
5.检查键是否存在if ('a', 'b') in my_dict_with_tuple.keys(): print("键 ('a', 'b') 存在")else: print("键 ('a', 'b') 不存在")
6.检查值是否存在if 3 in my_dict_with_tuple.values(): print("值 3 存在")else: print("值 3 不存在")2.2.6 bytes 与 bytearray
作用:bytes 和 bytearray 都是用来表示二进制数据的类型,但它们有重要的区别。bytes 是不可变的,而 bytearray 是可变的。它们都用于处理二进制数据,例如文件读写、网络通信等场景。
| 特性 | bytes | bytearray |
|---|---|---|
| 可变性 | ❌ 不可变 | ✅ 可变 |
| 内存 | 创建后固定 | 可动态修改 |
| 适用场景 | 常量、哈希键、网络传输 | 需要频繁修改的二进制数据 |
1.常见 bytes 操作# ========== 1. 创建 bytes ==========b1 = b'hello' # 字面量b2 = bytes('你好', 'utf-8') # 字符串编码b3 = bytes([72, 101, 108]) # 从整数列表创建 → b'Hel'b4 = bytes(5) # 5个零字节 → b'\x00\x00\x00\x00\x00'
# ========== 2. 索引与切片(与 str 类似)==========b = b'Hello, World!'b[0] # 72(返回整数,不是 bytes!)b[0:5] # b'Hello'b[-1] # 33('!' 的 ASCII 值)
# ========== 3. 常用方法 ==========b = b' Hello, World! 'b.strip() # b'Hello, World!'b.lower() # b' hello, world! 'b.split(b',') # [b' Hello', b' World! ']b.replace(b'World', b'Python') # b' Hello, Python! 'b.startswith(b' He') # True
# ========== 4. 编码与解码 ==========text = '你好世界'encoded = text.encode('utf-8') # str → bytesdecoded = encoded.decode('utf-8') # bytes → str → '你好世界'
# ========== 5. 拼接与重复 ==========b1 = b'Hello'b2 = b' World'b1 + b2 # b'Hello World'b'Ha' * 3 # b'HaHaHa'
# ========== 6. 判断与查找 ==========b = b'abcdef'b'cd' in b # Trueb.index(b'd') # 3b.count(b'a') # 1
# ========== 7. 与整数互转 ==========b = b'\x00\x01\x02\x03'list(b) # [0, 1, 2, 3]bytes([255, 128, 0]) # b'\xff\x80\x00'
# ========== 8. hex / fromhex ==========b = b'\xde\xad\xbe\xef'b.hex() # 'deadbeef'bytes.fromhex('deadbeef') # b'\xde\xad\xbe\xef'
#bytearray 的可变操作ba = bytearray(b'Hello')
2. ✅ 可变操作(bytes 不支持)ba[0] = 72 # 按索引修改(赋值整数)ba[1:3] = b'EL' # 切片赋值ba.append(33) # 追加字节 → bytearray(b'HELLO!')ba.extend(b' World') # 扩展ba.insert(5, 44) # 插入ba.pop() # 弹出最后一个字节del ba[5] # 删除指定位置
# ⚠️ 注意:ba[i] = 整数,ba[i:j] = bytes/bytearrayba[0] = ord('h') # ✅ 正确# ba[0] = b'h' # ❌ TypeError!
3. 典型应用场景# 📡 网络通信:读取 socket 数据data = sock.recv(1024) # 返回 bytesheader = data[:4] # 解析头部
# 📁 文件读写(二进制模式)with open('image.png', 'rb') as f: content = f.read() # bytes
# 🔧 协议解析/修改(需要可变)buf = bytearray(header + payload)buf[2] = 0x01 # 修改标志位buf[4:8] = length.to_bytes(4, 'big') # 更新长度字段sock.sendall(buf)| 特性 | Python | Java | C++ |
|---|---|---|---|
| 不可变二进制 | bytes | byte[] (约定俗成) / String (UTF-8) | const uint8_t* / std::string |
| 可变二进制 | bytearray | byte[] (原生可变) | uint8_t[] / std::vector<uint8_t> |
| 底层数据单元 | 整数 (0-255) | byte 类型 (有符号, -128~127) | char / uint8_t (无符号, 0-255) |
| 内存管理 | 自动垃圾回收 (GC) | 自动垃圾回收 (GC) | 手动管理 (new/delete) 或 RAII (std::vector) |
| 文件读写模式 | open('f', 'rb') 返回 bytes | FileInputStream 读入 byte[] | std::ifstream 配合 std::ios::binary |
| 与文本的界限 | 极其严格 (str 与 bytes 必须通过 .encode()/.decode() 转换) | 相对模糊 (byte[] 与 String 转换需指定字符集) | 完全没有界限 (一切皆内存字节,文本和二进制无类型区分) |
| 典型应用场景 | 网络协议解析、加密、文件处理 | 网络 I/O、序列化、加密算法 | 底层驱动、高性能网络服务器、游戏引擎、音视频处理 |
2.3数据类型转换
2.3.1 隐式类型转换
即自动类型转换,Python 会根据操作数的类型自动进行类型提升。例如,当一个整数与一个浮点数进行运算时,整数会被自动转换为浮点数。
2.3.2 显式类型转换
常用的显式类型转换函数有:
int(x):将 x 转换为整数类型。float(x):将 x 转换为浮点数类型。str(x):将 x 转换为字符串类型。list(x):将 x 转换为列表类型。tuple(x):将 x 转换为元组类型。set(x):将 x 转换为集合类型。dict(x):将 x 转换为字典类型(x 必须是可迭代的键值对)。bytes(x):将 x 转换为 bytes 类型。bytearray(x):将 x 转换为 bytearray 类型。
下面的内容适合作为课程讲义,默认学生已经具备 C++、Java 基础,因此重点突出 Python 与 C++/Java 的区别,并给出最常用的写法和示例。
2.4 分支、循环、函数
Python 的程序控制结构与 C++、Java 基本一致,都包括顺序、分支、循环三种结构。
但 Python 的语法更加简洁,最大的特点是使用缩进表示代码块,而不是 {}。
2.4.1 分支语句
(1)if 语句
Python 使用 if、elif、else 实现条件判断。
语法格式:
if 条件: 语句块1elif 条件: 语句块2else: 语句块3注意:
- 条件后必须加
:- 使用缩进表示代码块(通常4个空格)
- 不需要使用大括号
{}
(2)Python 特点
Python 可以直接进行链式比较:
age = 20
if 18 <= age < 60: print("成年人")相比 C++:
if(age >=18 && age <60)Python 写法更加自然、简洁。
2.4.2 循环语句
Python 提供两种循环:
whilefor
其中 for 循环使用频率远高于 while。
(1)while 循环
语法:
while 条件: 循环体示例:
i = 1
while i <= 5: print(i) i += 1(2)for 循环
Python 的 for 与 C++ 不同。
C++:
for(int i=0;i<5;i++)Python:
for i in range(5): print(i)Python 的 for 实际上是遍历一个可迭代对象。
(3)range() 函数
range() 函数用于生成一个整数序列,常用于控制循环次数。
最常用的三种形式:
range(stop) #此处 stop 表示生成的整数序列的结束值(不包含 stop),默认从 0 开始,步长为 1。for i in range(5): print(i)结果:
0 1 2 3 4range(start, stop)for i in range(2, 6): print(i)结果:
2 3 4 5range(start, stop, step) #开始,截至,步数for i in range(0, 10, 2): print(i)结果:
0 2 4 6 8注意:当 step=-1时意味着倒序遍历,即从右向左遍历。
(4)遍历列表
Python 最常见的写法:
names = ["Tom", "Alice", "Bob"]
for name in names: print(name)输出:
TomAliceBob相比 C++ 使用下标遍历更加简洁。
(5)break 与 continue
与 C++、Java 用法基本一致。
break:结束整个循环
for i in range(10): if i == 5: break print(i)输出:
01234continue:跳过本次循环
for i in range(5): if i == 2: continue print(i)输出:
01342.4.3 可更改对象与不可更改对象
在 Python 中,str、tuple 和数字是不可更改的对象,而 list、dict 等则是可以修改的对象。
不可变类型:变量赋值 a = 5 后再赋值 a = 10,这里实际是新生成一个 int 值对象 10,再让 a 指向它,而 5 被丢弃。这个过程不是改变 a 的值,而是让 a 指向了新对象。
可变类型:变量赋值 la = [1, 2, 3, 4] 后再赋值 la[2] = 5,则是将列表 la 的第三个元素值更改。列表本身没有改变,只是其内部的一部分值被修改了。
Python 函数的参数传递:
不可变类型:类似 C++ 的值传递,如整数、字符串、元组。如 fun(a),函数内部重新绑定 a 不会影响函数外部的对象。
可变类型:类似 C++ 的引用传递,如列表、字典。如 fun(la),在函数内部修改列表内容后,函数外部的 la 也会受到影响。
Python 中一切都是对象。严格来说,不能简单地说是值传递还是引用传递,更准确的说法是:传递不可变对象或可变对象。
2.4.4 函数定义与调用
函数用于封装可重复使用的代码,提高程序的可读性和复用性。
Python 使用 def 定义函数。
(1)定义函数
语法:
def 函数名(参数): 函数体例如:
def hello(): print("Hello Python")调用:
hello()输出:
Hello Python(2)带参数函数
def add(a, b): return a + b
result = add(3, 5)
print(result)输出:
8(3)默认参数
Python 支持默认参数,这是 C++ 常用但 Java 不支持的特性。
def greet(name="Python"): print("Hello,", name)
greet()
greet("Alice")输出:
Hello, PythonHello, Alice(4)返回多个值
Python 可以一次返回多个值,本质上返回的是元组(Tuple)。
def calculate(a, b): return a + b, a - b
x, y = calculate(10, 3)
print(x)print(y)输出:
137相比 C++,无需结构体或引用参数即可返回多个结果。
(5)关键字参数
Python 可以指定参数名称,提高代码可读性。
def student(name, age): print(name, age)
student(age=20, name="Tom")输出:
Tom 20参数顺序可以改变,只需保证参数名正确。
(6)可变参数(了解)
当参数个数不确定时,可以使用 *args。
def total(*args): print(sum(args))
total(1, 2, 3)
total(10, 20, 30, 40)输出:
6100小结
| 内容 | Python 特点 | 与 C++/Java 对比 |
|---|---|---|
| if 分支 | 使用缩进表示代码块 | 不使用 {} |
| for 循环 | for...in 遍历对象 | 不采用 for(;;) 形式 |
| range() | 快速生成整数序列 | 常用于控制循环次数 |
| while | 与 C++ 基本一致 | 语法更简洁 |
| 函数定义 | 使用 def | 无需声明返回类型 |
| 默认参数 | 支持 | Java 不支持 |
| 多返回值 | 可直接返回多个值 | C++/Java 实现较复杂 |
| 关键字参数 | 支持按参数名传参 | C++/Java 不支持 |
| 可变参数 | *args | 类似 C++ 可变参数,但更安全、更易用 |
2.5 lambda 表达式
Python 使用 lambda 创建匿名函数。
[!重要] lambda 函数可以有任意数量的参数,但只能包含一个表达式,并返回该表达式的结果。
2.5.1 语法格式
lambda arguments: expression- lambda 是 Python 的关键字,用于定义 lambda 函数。
- arguments 是参数列表,可以包含零个或多个参数,需写在冒号(
:)前。 - expression 是用于计算并返回结果的表达式。
示例:
square = lambda x: x ** 2print(square(5)) # 252.5.2 lambda 函数的特点
- 匿名:没有函数名称,可以赋值给变量或作为参数传递给其他函数。
- 简洁:通常只包含一行代码,适合编写简单函数。
- 常见用途:作为
map()、filter()、reduce()等函数的参数,将简单的处理逻辑直接传入这些函数。
例如,使用 lambda 作为 map() (映射)的参数,对列表中的每个元素求平方:
numbers = [1, 2, 3, 4]squares = list(map(lambda x: x ** 2, numbers)) # map() 函数会将 lambda 函数应用到 numbers 列表的每个元素上,并返回一个迭代器,最后使用 list() 将其转换为列表。print(squares) # [1, 4, 9, 16]使用 lambda 作为 filter()(过滤)的参数,只保留列表中的偶数:
numbers = [1, 2, 3, 4, 5, 6]even_numbers = list(filter(lambda x: x % 2 == 0, numbers))print(even_numbers) # [2, 4, 6]使用 lambda 作为 reduce() (计算总和)的参数,计算列表中所有元素的和。使用前需要从 functools 导入 reduce:
from functools import reduce
numbers = [1, 2, 3, 4]total = reduce(lambda x, y: x + y, numbers)print(total) # 102.5.3 lambda 函数的综合使用
除了 map()、filter() 和 reduce(),lambda 还常用于 sorted() 的 key 参数中,根据指定规则排序。下面综合使用这些函数,先筛选出及格成绩,再按成绩从高到低排序,最后计算及格学生的总分:
from functools import reduce
students = [ {"name": "小明", "score": 85}, {"name": "小红", "score": 92}, {"name": "小刚", "score": 58}, {"name": "小丽", "score": 76},]
# filter():筛选出及格的学生#students是一个列表,filter()函数会将lambda函数应用到students列表的每个元素上,并返回一个迭代器,最后使用list()将其转换为列表。#lambda student: student["score"] >= 60是一个匿名函数,接收一个学生字典作为参数,返回该学生的成绩是否大于等于60。passed = list(filter(lambda student: student["score"] >= 60, students))
# sorted():按成绩从高到低排序ranked = sorted(passed, key=lambda student: student["score"], reverse=True) #reverse=True表示降序排列
# map():提取排序后的学生姓名names = list(map(lambda student: student["name"], ranked))
# reduce():计算及格学生的总分total_score = reduce( lambda total, student: total + student["score"], passed, 0,) # reduce()三个参数,第一个参数是函数,第二个参数是可迭代对象,第三个参数是初始值,这里初始值为0,表示从0开始累加。
print(names) # ['小红', '小明', '小丽']print(total_score) # 2532.5.4 推导式
推导式可以用简洁的语法,根据一个可迭代对象快速创建新的序列。常见类型包括:
- 列表(
list)推导式 - 字典(
dict)推导式 - 集合(
set)推导式 - 生成器(
generator)表达式
列表推导式
列表推导式的基本格式为:
[表达式 for 变量 in 可迭代对象]也可以添加条件,只保留符合条件的元素:
[表达式 for 变量 in 可迭代对象 if 条件]其中,表达式用于生成列表元素,for用于遍历可迭代对象,if用于过滤元素。
1. 生成指定范围的数字
squares = [number ** 2 for number in range(1, 6)]#其中,`number ** 2` 是表达式,表示将 `number` 的平方作为列表元素;`for number in range(1, 6)` 表示遍历从 1 到 5 的整数。print(squares) # [1, 4, 9, 16, 25]2. 过滤元素
even_numbers = [number for number in range(10) if number % 2 == 0]#其中,`number for number in range(10)` 表示遍历从 0 到 9 的整数;`if number % 2 == 0` 表示只保留偶数。print(even_numbers) # [0, 2, 4, 6, 8]3. 转换字符串
names = ["alice", "bob", "carol"]upper_names = [name.upper() for name in names]print(upper_names) # ['ALICE', 'BOB', 'CAROL']4. 使用条件表达式
scores = [58, 72, 90]results = ["及格" if score >= 60 else "不及格" for score in scores]print(results) # ['不及格', '及格', '及格']5. 字典推导式
字典推导式使用 {键: 值 for 变量 in 可迭代对象} 创建字典:
numbers = [1, 2, 3]squares = {number: number ** 2 for number in numbers}print(squares) # {1: 1, 2: 4, 3: 9}6. 集合推导式
集合推导式使用 {表达式 for 变量 in 可迭代对象} 创建集合,并会自动去重:
words = ["Python", "python", "Java"]lower_words = {word.lower() for word in words}print(lower_words) # {'python', 'java'}7. 生成器表达式
生成器表达式使用圆括号,采用惰性计算,适合处理较大的数据:
numbers = (number ** 2 for number in range(1, 4))print(list(numbers)) # [1, 4, 9]2.6 python面向对象编程(OOP)
首先来了解一下,相比于 C++ 和 Java,Python 的面向对象编程具有以下特点:
| 特性 | Python | Java | C++ |
|---|---|---|---|
| 类型系统 | 动态类型 | 静态类型 | 静态类型 |
| 封装机制 | 下划线约定(非强制) | 关键字强制(private等) | 关键字强制(含友元) |
| 多态实现 | 鸭子类型(天然动态) | 接口/抽象类 | 显式虚函数(virtual) |
| 继承方式 | 多重继承(C3算法) | 单继承 + 多接口 | 多重继承(含虚继承) |
| 内存管理 | 自动垃圾回收(GC) | 自动垃圾回收(GC) | 手动管理 / 智能指针 |
| 当前对象引用 | 显式 self | 隐式 this | 隐式 this 指针 |
- 一切皆对象:整数、字符串、函数和类本身都是对象。
- 动态类型:定义变量时无需声明类型,类型在运行时确定。
- 鸭子类型:只要对象支持所需的操作,就可以被使用,无需显式继承特定类。
2.6.1 类的定义与实例化
使用 class 关键字定义类,构造方法 __init__ 用于初始化实例属性,实例方法的第一个参数通常为 self:
注意:
1. 在Python中,self是实例方法的第一个参数,它代表了类的实例本身,而非类,相当于Java中的this关键字。 2. __init__是类的构造方法,用于初始化实例属性。 3. 在类的内部,使用 def 关键字来定义一个方法,与一般函数定义不同,类方法必须包含参数 self。
基本语法格式如下:
class 类名: 类属性 = 属性值
def __init__(self, 参数1, 参数2): self.实例属性1 = 参数1 self.实例属性2 = 参数2
def 方法名(self, 参数): # 方法体 return 结果
对象 = 类名(参数1, 参数2)对象.方法名(参数)class Person: def __init__(self, name, age): self.name = name self.age = age
def introduce(self): return f"我叫{self.name},今年{self.age}岁"
# 实例化对象并调用方法person = Person("小明", 18)print(person.name) # 小明print(person.introduce()) # 我叫小明,今年18岁类属性由所有实例共享,实例属性则属于具体对象:
class Dog: species = "犬科" # 类属性
def __init__(self, name): self.name = name # 实例属性
def bark(self): print(f"{self.name}:汪汪!")
dog = Dog("旺财")print(Dog.species) # 犬科dog.bark() # 旺财:汪汪!2.6.2 继承与多态
继承可以让子类复用父类的属性和方法,并根据需要扩展或重写行为。多态表示同一个操作作用于不同对象时,可以产生不同的结果。
Python、Java 与 C++ 的继承和多态对比
| 对比项 | Python | Java | C++ |
|---|---|---|---|
| 类型系统 | 动态类型 | 静态类型 | 静态类型 |
| 类继承 | 支持多继承 | 类只支持单继承,可实现多个接口 | 支持多继承 |
| 方法重写 | 直接在子类中定义同名方法,可使用 super() | 使用 @Override,通常依赖接口或父类 | 通常使用 virtual 与 override |
| 方法解析 | 按 MRO(方法解析顺序)查找方法 | 沿单一父类链查找 | 受继承方式、虚函数和虚继承影响 |
| 多态基础 | 鸭子类型:关注对象能做什么 | 关注对象声明的父类或接口类型 | 关注继承关系和虚函数 |
| 是否必须显式继承统一接口 | 通常不需要 | 通常需要 | 通常需要 |
1. 多继承与 MRO
一个 Python 类可以同时继承多个父类。当多个父类具有同名方法时,Python 会按照 MRO 决定调用顺序。可以通过 类名.mro() 查看该顺序。
经典的菱形继承示例如下:
class Animal: def speak(self): print("动物发出声音")
class Flyable(Animal): def speak(self): print("会飞的动物发出声音")
class Swimmable(Animal): def speak(self): print("会游泳的动物发出声音")
class Duck(Flyable, Swimmable): pass
duck = Duck()duck.speak() # 会飞的动物发出声音
print([cls.__name__ for cls in Duck.mro()])# ['Duck', 'Flyable', 'Swimmable', 'Animal', 'object']Duck 先继承 Flyable,因此会优先找到 Flyable.speak()。在实际项目中,多继承常与 super() 配合使用;参与协作的类应采用一致的方法签名,并继续调用 super(),让 MRO 链上的实现都有机会执行。
2. 鸭子类型与多态
Python 的多态通常不要求对象继承同一个父类。只要对象提供了所需的方法,就可以被同一段代码使用,这就是“鸭子类型”。
class Dog: def speak(self): return "汪汪"
class Cat: def speak(self): return "喵喵"
class Robot: def speak(self): return "你好,我是机器人"
def make_it_speak(obj): print(obj.speak())
for speaker in (Dog(), Cat(), Robot()): make_it_speak(speaker)Dog、Cat 和 Robot 没有共同的自定义父类,但它们都实现了 speak(),因此 make_it_speak() 可以统一处理它们。与 Java 接口或 C++ 虚函数相比,这种方式更灵活,但错误通常要到运行时才会暴露。
注意:Python确实没有传统意义上的多态机制,它靠的是运行时名字查找。但“多态”作为一种编程现象(同一接口,不同行为)在Python中依然存在,只是实现路径完全不同。
2.6.3 封装与私有属性
封装是把数据和操作数据的方法组织在类中,并控制外部代码如何访问内部状态。Python 更强调约定和明确的接口,而不是强制禁止访问。
1. 属性的访问约定
| 写法 | 含义 | 典型用途 |
|---|---|---|
name | 公开属性 | 允许类内外直接访问 |
_name | 受保护属性(约定) | 提示外部代码和子类谨慎使用 |
__name | 触发名称改写 | 避免子类意外覆盖,表达类内部使用的意图 |
以两个下划线开头、结尾没有两个下划线的属性(如 __balance)会被 Python 改写为 _类名__属性名。因此它不能通过 对象.__balance 直接访问,但这并不是绝对的安全或权限机制,不应依赖它保存敏感信息。
class BankAccount: def __init__(self, owner, balance=0): self.owner = owner self.__balance = balance
def deposit(self, amount): if amount <= 0: raise ValueError("存款金额必须大于 0") self.__balance += amount
def get_balance(self): return self.__balance
def __check_password(self, password): return password == "123456"
def withdraw(self, amount, password): if not self.__check_password(password): raise PermissionError("密码错误") if amount > self.__balance: raise ValueError("余额不足") self.__balance -= amount
account = BankAccount("小明", 1000)account.deposit(500)account.withdraw(200, "123456")print(account.get_balance()) # 1300
# print(account.__balance) # AttributeError# account.__check_password("123456") # AttributeError示例中的 __balance 是私有属性,__check_password() 是私有方法。类的内部通过 self.__balance 和 self.__check_password() 使用它们,外部则通过公开方法与对象交互。
2. 实例方法与 self
类中使用 def 定义实例方法时,第一个参数用于接收当前实例,按照约定命名为 self。调用 account.deposit(500) 时,Python 会自动把 account 作为 self 传入。
self 不是关键字,理论上可以换成其他名称,但统一使用 self 能让代码更容易阅读和维护。
3. 专有方法
以双下划线开头和结尾的方法称为专有方法(也常称“魔术方法”或 dunder 方法)。它们定义对象如何响应 Python 的内置操作,通常由解释器间接调用。
| 专有方法 | 触发方式 | 作用 |
|---|---|---|
__init__ | ClassName(...) | 初始化新创建的实例 |
__del__ | 对象被回收时 | 执行清理逻辑,但调用时机不确定,不适合管理关键资源 |
__repr__ | repr(obj)、交互式环境 | 返回供开发者查看的对象表示 |
__setitem__ | obj[key] = value | 按键或索引赋值 |
__getitem__ | obj[key] | 按键或索引取值 |
__len__ | len(obj) | 返回对象长度 |
__call__ | obj(...) | 让实例可以像函数一样调用 |
__add__ | obj + other | 定义加法 |
__sub__ | obj - other | 定义减法 |
__mul__ | obj * other | 定义乘法 |
__truediv__ | obj / other | 定义真除法 |
__mod__ | obj % other | 定义取余运算 |
__pow__ | obj ** other | 定义乘方运算 |
Python 3 不再使用 __cmp__。比较操作应分别实现 __eq__、__lt__、__le__、__gt__、__ge__ 等方法。
下面的类通过专有方法表现得像一个内置容器:
class ScoreBook: def __init__(self): self.__scores = {}
def __setitem__(self, name, score): if not 0 <= score <= 100: raise ValueError("分数必须在 0 到 100 之间") self.__scores[name] = score
def __getitem__(self, name): return self.__scores[name]
def __len__(self): return len(self.__scores)
def __repr__(self): return f"ScoreBook({self.__scores!r})"
scores = ScoreBook()scores["小明"] = 95scores["小红"] = 88
print(scores["小明"]) # 95print(len(scores)) # 2print(scores) # ScoreBook({'小明': 95, '小红': 88})3. Python 高级进阶
这一章围绕一个常见需求展开:创建独立开发环境,编写采集或处理数据的脚本,让程序运行得更稳定,再把结果保存下来。每节的完整 Python 示例都可单独运行;需要其他文件或第三方依赖时会提前说明。
示例使用 Python 3.11+,Windows 命令默认在 PowerShell 中执行,Linux/macOS 的差异单独列出。先通过 python --version 确认版本;安装依赖前,应激活自己项目的环境。文中的安装和联网命令供读者按需执行。
3.1 Python 虚拟环境
3.1.1 为什么每个项目需要独立环境
假设旧项目依赖某个库的 1.x 版本,新项目需要 2.x。如果全部安装到同一个 Python 环境,升级新项目可能使旧项目无法启动。虚拟环境为每个项目提供独立的依赖目录,让安装、升级、重建都能围绕项目进行。

图中的 Django 版本仅用于解释冲突,并非新项目的版本推荐。“隔离”主要指 Python 依赖;虚拟环境不隔离文件访问、网络、操作系统或 GPU 驱动,也不是运行不可信代码的安全沙箱。
3.1.2 venv 与 Conda 如何选择
| 对比项 | venv | Conda |
|---|---|---|
| 来源 | Python 标准库自带 | 需要安装 Conda 发行版 |
| Python 版本 | 基于创建环境时已有的解释器 | 可以在环境中安装指定版本解释器 |
| 管理范围 | Python 环境,通常搭配 pip 安装包 | Python 包以及渠道提供的非 Python 库、工具 |
| 典型场景 | Web 服务、自动化脚本、普通 Python 项目 | 科研、数据分析、依赖复杂二进制库的项目 |
| 共享方式 | 依赖清单或项目锁文件 | environment.yml,必要时加平台专用锁定信息 |
| 注意事项 | 不要把 .venv 复制到另一台机器 | 渠道、操作系统和驱动仍影响可复现性 |
普通脚本从 venv + pip 入门即可;团队已有 Conda 环境文件时,优先沿用。Conda 能安装部分 CUDA 运行库,但不能代替宿主机显卡驱动,深度学习环境仍需核对框架与驱动兼容性。
3.1.3 创建、激活和确认 venv
在项目目录执行:
python -m venv .venv.\.venv\Scripts\Activate.ps1python -c "import sys; print(sys.executable)"python -m pip --versionLinux/macOS 的创建与激活命令如下:
python3 -m venv .venvsource .venv/bin/activatepython -c "import sys; print(sys.executable)"输出的解释器路径应位于项目的 .venv 中。工作结束后执行 deactivate 退出;把 .venv/ 加入 .gitignore,提交源代码和依赖清单即可。
如果 PowerShell 阻止激活脚本,可以直接调用环境中的解释器,无需修改执行策略:
.\.venv\Scripts\python.exe -m pip --version.\.venv\Scripts\python.exe main.py第二条命令要求项目中已有 main.py。在 VS Code 中通过 Python: Select Interpreter 选择同一解释器,再新建终端。判断环境是否正确应看 sys.executable,不能只看终端前面的环境名。venv 官方说明
3.1.4 Conda 常用工作流
下面假设 Conda 已安装且当前终端可以执行 conda:
conda create -n data-lab python=3.12conda activate data-labconda install numpy pandasconda env listpython -c "import sys; print(sys.executable)"conda env export --from-history > environment.ymlconda deactivate另一台机器使用 conda env create -f environment.yml 重建环境。--from-history 主要记录主动安装的 Conda 依赖,便于跨平台调整,但不锁定全部传递依赖,也不完整记录 pip 安装项;需要精确复现实验时,还应记录完整版本、平台和 pip 依赖。
优先安装 Conda 包,再在已激活的环境中用 python -m pip 补充渠道中没有的包。混用后避免反复让两个工具替换同一组依赖,环境冲突严重时从清单重建通常更清晰。Conda 环境管理
3.2 Python 的包管理工具
3.2.1 环境、依赖声明和锁定的区别
| 工具或文件 | 解决的问题 | 实践建议 |
|---|---|---|
| pip | 给指定 Python 安装、卸载发行包 | 使用 python -m pip 避免装到其他解释器 |
requirements.txt | 批量安装一组依赖 | 可以指定版本,也可以引用带哈希的固定依赖清单 |
pyproject.toml | 声明项目元数据、依赖和工具配置 | 用于可维护的项目,不只是一次性脚本 |
| uv | 环境、依赖解析、锁定与项目运行 | 团队采用后统一使用其工作流 |
| Poetry | 项目依赖、锁定、构建和发布 | 已有 Poetry 项目继续使用原有配置 |
| pipx | 为独立命令行工具提供隔离环境 | 适合安装工具,不负责项目运行依赖 |
依赖声明描述“允许使用哪些版本”,锁文件记录解析出的具体依赖。锁文件仍不能保证不同操作系统、CPU 架构和外部服务完全一致。
3.2.2 pip 的日常命令
以下操作在虚拟环境中执行:
python -m pip install requestspython -m pip show requestspython -m pip listpython -m pip checkpython -m pip freeze > requirements.txt重建环境时使用 python -m pip install -r requirements.txt;主动升级用 python -m pip install --upgrade requests,卸载用 python -m pip uninstall requests。升级后应运行相关测试,再更新依赖清单。
pip freeze 输出当前环境中已安装包的清单,不会为项目生成完整的跨平台锁定方案;请在干净的项目环境中导出。Windows PowerShell 5.1 的 > 可能写成 UTF-16,可改为:
python -m pip freeze | Out-File -Encoding utf8 requirements.txt包的发行名称与导入名称也可能不同,例如安装 beautifulsoup4,代码中却是 from bs4 import BeautifulSoup。遇到 ModuleNotFoundError,先比对解释器路径与 python -m pip show 包名 的安装位置。pip 用户指南
3.2.3 一个 uv 项目工作流
以下命令要求已安装 uv,并且在准备存放项目的父目录执行:
uv init collector-democd collector-demouv add requestsuv add --dev pytestuv run python -c "import requests; print(requests.__version__)"uv sync --lockeduv add 更新项目依赖与锁文件;uv run 在项目环境中执行命令;uv sync --locked 在锁文件与声明不一致时失败,适合 CI 检查。提交 pyproject.toml 和 uv.lock,让协作者使用同一套依赖解析结果。uv 项目指南
3.2.4 安装失败时检查什么
| 现象 | 优先检查 |
|---|---|
| 已安装却无法导入 | 当前解释器、虚拟环境、发行名与导入名 |
No matching distribution found | Python 版本、平台、架构、包名和配置的索引 |
| 构建 wheel 失败 | 是否缺少编译器或系统库,是否有适配的预编译包 |
| 依赖版本冲突 | 阅读冲突链,调整直接依赖约束,而非强行忽略依赖 |
| 下载超时 | 网络、代理、组织指定的软件源 |
排障时保留最早的有效错误信息。不要为了安装成功随意关闭证书校验,也不要把包含令牌的私有源地址提交到仓库。
3.3 Python 的性能优化
3.3.1 先定位慢在哪里
处理一份大日志时,瓶颈可能是磁盘、字符串解析、重复查找,也可能是网络等待。先记录输入规模、Python 版本、总耗时和峰值内存,再决定优化方法。
| 工具 | 适合测量 | 注意事项 |
|---|---|---|
time.perf_counter() | 整段业务耗时 | 包含等待时间,适合比较端到端体验 |
timeit | 小段纯计算代码 | 重复测量,避免把初始化混入比较 |
cProfile | 函数调用次数和累计耗时 | 找热点函数,而非凭感觉改代码 |
tracemalloc | Python 内存分配 | 不等于整个进程内存,也不覆盖所有原生库分配 |
如果已有 report.py,可执行:
python -m cProfile -s cumulative report.py先看累计耗时高且调用频繁的函数。把计算移到另一行、把循环改成推导式,都不一定改变真正的瓶颈。
3.3.2 实战:用集合过滤重复编号
场景:从一批编号中筛选未处理的数据。列表成员查找通常是 O(n),集合成员查找平均为 O(1),大批量判断时差异很明显。
from timeit import repeat
processed = list(range(10_000))incoming = list(range(9_000, 11_000))
def filter_with_list(): return [item for item in incoming if item not in processed]
def filter_with_set(): processed_set = set(processed) return [item for item in incoming if item not in processed_set]
assert filter_with_list() == filter_with_set()for func in (filter_with_list, filter_with_set): samples = repeat(func, number=3, repeat=3) print(func.__name__, f"{min(samples) / 3:.6f} 秒/次")此处将建集合的成本也计入测量。若同一个集合能用于多批任务,可把构建移到批次外。集合会占用额外内存,元素必须可哈希;它适合判断是否存在,不适合保留重复次数。输出耗时随硬件变化,应比较相同环境下的结果。
3.3.3 实战:逐行处理大文件
不需要一次性拿到全部内容时,用迭代器避免 read() 后再 splitlines() 创建大对象:
from pathlib import Pathfrom tempfile import TemporaryDirectory
def iter_errors(path): with path.open(encoding="utf-8") as file: for number, line in enumerate(file, start=1): if "ERROR" in line: yield number, line.rstrip("\n")
with TemporaryDirectory() as directory: path = Path(directory) / "app.log" path.write_text("INFO start\nERROR timeout\nINFO done\n", encoding="utf-8") for number, message in iter_errors(path): print(number, message) # 2 ERROR timeoutyield 让调用方按需取数据。内存主要取决于当前行和下游是否累积结果;如果立刻调用 list(iter_errors(path)),仍会保存所有匹配项。
3.3.4 缓存、批处理与优化边界
重复执行昂贵的纯函数时,可以考虑 functools.lru_cache(maxsize=...);参数需可哈希,结果不能依赖随时变化的外部状态。数据库查询和 HTTP 请求更适合先减少次数、批量处理、复用连接,再考虑增加并发。
优化后至少确认结果相同,并比较代表性小输入和大输入。不要给未设失效策略的缓存无限增长空间,也不要假设换成线程就能加速所有 Python 计算。
3.4 Python 的多线程与异步编程
3.4.1 根据任务选择并发模型
| 工作类型 | 常用方案 | 典型场景 |
|---|---|---|
| 阻塞 I/O,已有同步库 | ThreadPoolExecutor | 多文件读取、同步 HTTP 客户端 |
| 大量等待,使用异步库 | asyncio | 多个 API 请求、长连接服务 |
| 纯 Python CPU 密集计算 | ProcessPoolExecutor | 独立的文本分析、计算任务 |
| 小任务、数量很少 | 顺序执行 | 并发开销可能超过收益 |
常规启用 GIL 的 CPython 中,同一进程通常不能由多个线程同时执行 Python 字节码,但 I/O 等待时可释放 GIL。某些原生扩展也会释放 GIL。Python 3.13 起提供可选的 free-threaded 构建,不能据此假定日常安装的解释器已关闭 GIL;应根据实际构建和依赖测量。线程与 GIL 说明
3.4.2 线程池:并发执行阻塞任务
下面用短暂休眠模拟三个耗时不确定的 I/O 任务,不需要外网。其中一个任务故意失败,演示如何保留其他任务的结果。
from concurrent.futures import ThreadPoolExecutor, as_completedfrom time import sleep
def read_record(record_id): sleep(0.02) if record_id == 2: raise ValueError("记录格式无效") return {"id": record_id, "status": "ok"}
with ThreadPoolExecutor(max_workers=3) as pool: futures = {pool.submit(read_record, number): number for number in range(3)} for future in as_completed(futures): record_id = futures[future] try: print("成功", future.result()) except ValueError as exc: print("失败", record_id, str(exc))结果的打印顺序不保证一致。限制 max_workers 可以控制同时运行的任务数,但一次提交百万个任务仍会占用大量内存,应分批提交或使用有界队列。
尽量让工作线程返回结果,由主线程合并。多个线程共同修改计数器等共享状态时,要使用 threading.Lock 或线程安全队列;不能依靠 GIL 保证一整段业务操作的原子性。
3.4.3 asyncio:限并发、超时与结果汇总
async def 定义协程函数;调用它只会产生协程对象,需要 await 或交给事件循环执行。下面模拟四个请求,同时最多执行两个,其中一个超时:
import asyncio
async def fetch_record(record_id, semaphore): async with semaphore: try: async with asyncio.timeout(0.08): delay = 0.15 if record_id == 3 else 0.01 await asyncio.sleep(delay) return {"id": record_id, "status": "ok"} except TimeoutError: return {"id": record_id, "status": "timeout"}
async def main(): semaphore = asyncio.Semaphore(2) async with asyncio.TaskGroup() as group: tasks = [ group.create_task(fetch_record(number, semaphore)) for number in range(4) ] print([task.result() for task in tasks])
if __name__ == "__main__": asyncio.run(main())结果按 tasks 列表顺序汇总:编号 0、1、2 为 ok,编号 3 为 timeout。这里的超时从拿到信号量后开始,不包括排队等待。TaskGroup 遇到未处理的普通异常时会取消其他任务,并在退出时报告异常组;示例把预期超时转换成结果,以便整批任务继续。asyncio 任务文档
在协程中直接执行 time.sleep() 或同步网络请求会阻塞事件循环。可使用对应异步库,或用 await asyncio.to_thread(blocking_function, ...) 迁移阻塞 I/O;取消等待不一定能停止后台线程,所以底层网络请求仍需自己的超时。
3.4.4 CPU 计算:进程池的入口保护
下面保存为 process_demo.py,使用 python process_demo.py 运行:
from concurrent.futures import ProcessPoolExecutor
def sum_squares(limit): return sum(number * number for number in range(limit))
if __name__ == "__main__": with ProcessPoolExecutor(max_workers=2) as pool: results = list(pool.map(sum_squares, [10_000, 20_000])) print(results) # [333283335000, 2666466670000]工作函数定义在模块顶层,参数与返回值需要可序列化。入口保护尤其关系到 Windows 的子进程启动;不要把这个例子直接粘进交互式控制台。这里只演示结构,小规模计算未必比单进程快。
3.5 Python 的装饰器
3.5.1 使用场景与执行顺序
多个函数都要统计耗时、检查权限或记录调用时,可以把公共行为写成装饰器。装饰器接收函数,返回包装后的可调用对象。
@decoratordef work(): ...上面是语法示意,等价于先定义 work,再执行 work = decorator(work)。装饰通常发生在函数定义时,包装函数的函数体则在每次调用时执行。
3.5.2 实战:保留函数信息的计时装饰器
from functools import wrapsfrom time import perf_counter
def timed(func): @wraps(func) def wrapper(*args, **kwargs): started = perf_counter() try: return func(*args, **kwargs) finally: elapsed = perf_counter() - started print(f"{func.__name__}: {elapsed:.6f}s") return wrapper
@timeddef total_price(prices, discount=1.0): """计算折扣后的总价。""" return sum(prices) * discount
print(total_price([10, 20], discount=0.9)) # 先输出耗时,再输出 27.0print(total_price.__name__) # total_price*args 和 **kwargs 转发参数,return 保留原函数返回值,finally 保证异常时也记录耗时,wraps 保留名称、文档和 __wrapped__ 等元数据。原函数异常仍会抛出,调用者可以正常处理。
3.5.3 实战:带参数的有限重试
重试适合短暂网络故障等可恢复错误,不能重试所有异常。下面只演示同步函数,且用本地模拟操作避免真实网络波动:
from functools import wrapsfrom time import sleep
def retry(*, attempts=3, delay=0.1, exceptions=(TimeoutError,)): if attempts < 1 or delay < 0: raise ValueError("attempts 至少为 1,delay 不能为负数")
def decorate(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(attempts): try: return func(*args, **kwargs) except exceptions: if attempt == attempts - 1: raise sleep(delay * (2 ** attempt)) return wrapper return decorate
def make_demo_reader(): calls = 0
@retry(attempts=3, delay=0) def read(): nonlocal calls calls += 1 if calls < 3: raise TimeoutError("暂时不可用") return "第三次读取成功"
return read
print(make_demo_reader()()) # 第三次读取成功生产环境还需总时限、抖动和服务端限流处理。对于扣款、创建订单等有副作用的请求,先设计幂等键或去重机制,避免重试造成重复操作。
@timed 放在 @retry(...) 上面时,计时覆盖整个重试过程;顺序颠倒时,每次尝试单独计时。装饰器从下往上应用。异步函数需要 async def wrapper 并 await func(...),否则统计到的可能只是创建协程对象的时间。
3.6 Python 的网络编程
3.6.1 日常开发优先理解 HTTP
调用开放 API 时,通常使用 HTTP 客户端,而不是直接手写 TCP 协议。一次请求包含方法、URL、请求头和可选请求体;响应包含状态码、响应头和响应体。
| 情况 | 含义 | 处理方式 |
|---|---|---|
| 2xx | 请求被成功处理 | 继续检查响应格式和业务字段 |
| 4xx | 请求、权限或资源问题 | 检查参数和授权,不盲目重试 |
| 429 | 触发限流 | 遵守 Retry-After,降低请求频率 |
| 5xx | 服务端处理失败 | 对适合重试的操作有限重试 |
| 超时、连接失败 | 可能没有收到有效 HTTP 响应 | 记录上下文,设置明确失败路径 |
3.6.2 实战:本地 JSON 服务与客户端
先把以下代码保存为 demo_server.py。它只监听本机回环地址,供后续章节练习;不用于生产部署。
import jsonfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom urllib.parse import parse_qs, urlsplit
class Handler(BaseHTTPRequestHandler): def do_GET(self): parsed = urlsplit(self.path) if parsed.path != "/api/hello": self.send_error(404, "Not Found") return name = parse_qs(parsed.query).get("name", ["访客"])[0] body = json.dumps({"message": f"你好,{name}"}, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
if __name__ == "__main__": with ThreadingHTTPServer(("127.0.0.1", 8765), Handler) as server: print("本地服务:http://127.0.0.1:8765,Ctrl+C 退出") try: server.serve_forever() except KeyboardInterrupt: pass在一个终端运行 python demo_server.py,再在另一个终端运行下面的 demo_client.py。如果 8765 已被占用,先选择空闲端口,并同步修改两端代码。
import jsonfrom urllib.error import HTTPError, URLErrorfrom urllib.parse import urlencodefrom urllib.request import Request, urlopen
query = urlencode({"name": "小明"})request = Request( f"http://127.0.0.1:8765/api/hello?{query}", headers={"Accept": "application/json"},)
try: with urlopen(request, timeout=3) as response: body = response.read(1_000_001) if len(body) > 1_000_000: raise ValueError("响应超过示例允许的大小") data = json.loads(body.decode("utf-8")) print(data["message"]) # 你好,小明except HTTPError as exc: print("HTTP 状态错误:", exc.code)except (URLError, TimeoutError) as exc: print("连接失败或超时:", exc)except (ValueError, KeyError, TypeError) as exc: print("响应内容不符合预期:", exc)urlencode() 负责查询参数转义;不要直接把包含空格、中文或 & 的值拼到 URL 中。响应通过上下文管理器关闭,读取设置了大小上限。标准库的 timeout 主要约束阻塞操作,不是整个业务流程的绝对总时限。
3.6.3 第三方客户端与连接复用
接口调用较多时,Requests 提供更简洁的参数与会话管理。先在当前环境执行 python -m pip install requests,并保持前面的本地服务运行:
import requests
try: with requests.Session() as session: response = session.get( "http://127.0.0.1:8765/api/hello", params={"name": "小明"}, timeout=(3, 5), ) response.raise_for_status() print(response.json()["message"])except requests.RequestException as exc: print("请求失败:", exc)except (ValueError, KeyError, TypeError) as exc: print("响应格式错误:", exc)timeout=(3, 5) 分别设置连接和读取超时,不等于请求总耗时最多 8 秒。Session 有助于复用连接。真实 HTTPS 请求保留默认的证书校验;凭据从环境变量或凭据系统读取,日志中不要打印授权头。Requests 使用说明
3.6.4 TCP 与 HTTP 的边界
TCP 提供有序字节流,没有消息边界:一次 send() 不一定对应一次 recv(),recv(1024) 也不保证收满 1024 字节。自定义协议需要长度前缀或分隔符、接收缓冲区、超时和消息大小限制。对接已有 HTTP API 时,把这些底层细节交给成熟客户端通常更合适。
3.7 JSON 与 Python 的交互
3.7.1 类型映射和四个常用函数
| JSON | Python | 需要注意 |
|---|---|---|
| object | dict | JSON 对象的键是字符串 |
| array | list | Python 元组序列化后再读取会变成列表 |
| string | str | JSON 字符串使用双引号 |
| number | int / float | 金额不要依赖二进制浮点精确计算 |
| true / false | True / False | 大小写不同 |
| null | None | 不等于空字符串或缺失字段 |
dumps() / loads() 在对象与字符串间转换,dump() / load() 面向文件对象。JSON 不支持注释、尾随逗号和 Python 的单引号字典写法。
import json
record = {"name": "小明", "active": True, "tags": ["Python", "数据处理"]}text = json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False)restored = json.loads(text)
print(text)assert restored == recordensure_ascii=False 保留可读中文,不决定文件编码;写文件时仍需指定 encoding="utf-8"。allow_nan=False 防止生成标准 JSON 不允许的 NaN、Infinity 等值。
3.7.2 实战:校验 API 返回的任务数据
“能解析成 JSON”只说明语法有效,不代表数据符合业务要求。下面的函数接收 JSON 字符串并检查必要字段:
import json
def parse_task(text): try: data = json.loads(text) except json.JSONDecodeError as exc: raise ValueError(f"JSON 语法错误,第 {exc.lineno} 行") from exc if not isinstance(data, dict): raise ValueError("任务必须是对象") if type(data.get("id")) is not int or data["id"] <= 0: raise ValueError("id 必须是正整数") title = data.get("title") if not isinstance(title, str) or not title.strip(): raise ValueError("title 必须是非空字符串") return {"id": data["id"], "title": title.strip()}
print(parse_task('{"id": 1, "title": " 整理日志 "}'))try: parse_task('{"id": true, "title": "错误示例"}')except ValueError as exc: print(exc) # id 必须是正整数这里刻意使用 type(...) is int,因为 Python 中 bool 是 int 的子类,而任务编号不应接受布尔值。不要使用 eval() 解析外部 JSON,它会执行 Python 表达式。
3.7.3 日期、金额和大批量记录
JSON 没有日期或 Decimal 类型。应约定明确的传输格式,例如带时区的 ISO 8601 字符串,以及以分计的整数金额:
import jsonfrom datetime import datetime, timezone
order = { "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc).isoformat(), "amount_cents": 1990, "currency": "CNY",}print(json.dumps(order, ensure_ascii=False, allow_nan=False))不建议用 default=str 随意转换所有未知对象,这容易把类型错误隐藏成字符串。金额若采用字符串形式,也要约定币种和小数位规则。
采集数据可用 JSON Lines(JSONL):每行一个完整 JSON 对象,便于追加和流式读取。不要连续向同一普通 JSON 文件调用多次 json.dump(),否则会产生相邻对象而非合法的单一 JSON 文档。
3.8 Python 的文件操作与数据持久化
3.8.1 先选择存储形式
| 需求 | 常用形式 | 局限 |
|---|---|---|
| 配置、少量嵌套记录 | JSON | 整体修改通常需要重写 |
| 规则表格、交给表格软件 | CSV | 类型和嵌套结构表达有限 |
| 逐条日志、采集结果 | JSONL | 多进程同时追加需要协调 |
| 本地查询、更新、事务 | SQLite | 同时写入的并发能力有限 |
| Python 对象临时缓存 | pickle | 不可加载不可信数据,不适合作为公共交换格式 |
3.8.2 路径、编码与文件模式
使用 pathlib.Path 组合路径。相对路径相对于进程工作目录,而不是自动相对于脚本;脚本旁文件可通过 Path(__file__).resolve().parent 定位,Notebook 中通常没有 __file__。
| 模式 | 行为 |
|---|---|
r | 读取,文件不存在时报错 |
w | 创建或清空后写入 |
a | 末尾追加 |
x | 仅创建新文件,已存在则报错 |
rb / wb | 读写二进制数据,不指定文本编码 |
下面用临时目录演示 JSONL,运行结束会清理示例数据:
import jsonfrom pathlib import Pathfrom tempfile import TemporaryDirectory
with TemporaryDirectory() as directory: path = Path(directory) / "records.jsonl" with path.open("w", encoding="utf-8") as file: for number in range(3): file.write(json.dumps({"id": number}, ensure_ascii=False) + "\n")
with path.open(encoding="utf-8") as file: for number, line in enumerate(file, start=1): try: print(json.loads(line)) except json.JSONDecodeError as exc: raise ValueError(f"第 {number} 行 JSON 无效") from exc写 CSV 时使用 newline="" 让 csv 模块处理换行;JSON 中的中文则同时需要正确的文件编码。文件句柄用 with 管理,异常时也能关闭。
3.8.3 实战:替换配置文件,避免留下半份 JSON
直接用 w 打开已有配置会先清空旧内容。如果写入中断,配置可能损坏。可以先在同目录写临时文件,成功后再替换目标:
import jsonimport osfrom pathlib import Pathfrom tempfile import NamedTemporaryFile, TemporaryDirectory
def save_json(path, data): path = Path(path) text = json.dumps(data, ensure_ascii=False, indent=2, allow_nan=False) path.parent.mkdir(parents=True, exist_ok=True) temporary_path = None try: with NamedTemporaryFile( mode="w", encoding="utf-8", dir=path.parent, delete=False ) as file: temporary_path = Path(file.name) file.write(text) file.flush() os.fsync(file.fileno()) os.replace(temporary_path, path) finally: if temporary_path is not None: temporary_path.unlink(missing_ok=True)
with TemporaryDirectory() as directory: path = Path(directory) / "settings.json" save_json(path, {"theme": "dark", "language": "zh-CN"}) print(json.loads(path.read_text(encoding="utf-8")))临时文件放在同目录以避免跨文件系统移动;先关闭文件也兼容 Windows。替换可以减少读取到半份文件的风险,但不等于多写入者事务,也不保证所有文件系统在掉电后的持久性。多个进程更新同一配置时仍需锁或数据库。
3.8.4 实战:用 SQLite 保存和更新采集结果
标准库 sqlite3 无需另装数据库服务,适合单机工具。示例数据库保存在临时目录;实际项目换成自己的数据路径即可。
import sqlite3from contextlib import closingfrom pathlib import Pathfrom tempfile import TemporaryDirectory
with TemporaryDirectory() as directory: path = Path(directory) / "articles.db" with closing(sqlite3.connect(path)) as connection: connection.row_factory = sqlite3.Row with connection: connection.execute( "CREATE TABLE articles (url TEXT PRIMARY KEY, title TEXT NOT NULL)" ) connection.executemany( "INSERT INTO articles (url, title) VALUES (?, ?) " "ON CONFLICT(url) DO UPDATE SET title = excluded.title", [ ("https://example.com/1", "第一篇"), ("https://example.com/2", "第二篇"), ("https://example.com/1", "第一篇(更新)"), ], ) rows = connection.execute("SELECT url, title FROM articles ORDER BY url") print([dict(row) for row in rows])? 占位符绑定数据,避免把外部值拼接进 SQL;它不能用于替换表名等 SQL 标识符。主键和 UPSERT 让重复采集更新已有记录。内层 with connection 成功时提交、异常时回滚,外层 closing 才负责关闭连接。sqlite3 连接上下文说明
并发程序最好把数据交给专门的写入线程或进程;不要随意跨线程共享连接。出现 database is locked 时先缩短事务、减少同时写入,再考虑更适合多用户服务的数据库。
3.9 Python 爬虫与 Scrapy 库
3.9.1 从任务规模选择工具
| 任务 | 优先考虑 |
|---|---|
| 网站有正式数据 API | 直接调用 API,校验结构化结果 |
| 少量静态页面 | HTTP 客户端加 HTML 解析库 |
| 多页面、分页、清洗、导出 | Scrapy |
| 内容必须执行 JavaScript 后才出现 | 先确认允许的接口,再评估浏览器自动化 |
先确认目标站点允许的访问方式、服务条款及 robots 规则,再限制频率和范围。Scrapy 默认下载 HTML,不会像浏览器一样自动执行页面中的 JavaScript。
3.9.2 建立可运行的练习项目
以下使用 Scrapy 官方教程中的练习站点 https://quotes.toscrape.com/。需要可访问外网,先在独立环境中安装 Scrapy:
python -m pip install scrapyscrapy startproject quote_labcd quote_lab生成目录的主要职责如下:
quote_lab/ scrapy.cfg quote_lab/ settings.py # 下载、并发与导出配置 items.py # 可选的数据字段定义 pipelines.py # 清洗和持久化 spiders/ __init__.py quotes.py # 接下来新建的爬虫3.9.3 提取内容并跟随分页
在 quote_lab/spiders/quotes.py 写入以下完整 Spider:
import scrapy
class QuotesSpider(scrapy.Spider): name = "quotes" allowed_domains = ["quotes.toscrape.com"] start_urls = ["https://quotes.toscrape.com/"]
def parse(self, response): for quote in response.css("div.quote"): yield { "text": quote.css("span.text::text").get(), "author": quote.css("small.author::text").get(), "tags": quote.css("div.tags a.tag::text").getall(), "source": response.url, } next_page = response.css("li.next a::attr(href)").get() if next_page: yield response.follow(next_page, callback=self.parse).get() 返回首个匹配或 None,.getall() 返回列表;response.follow() 根据当前页面解析相对链接。yield dict 产生数据,yield Request 安排后续请求,网络下载由 Scrapy 调度。Scrapy 官方教程
3.9.4 设置超时、限速和采集上限
在生成的 quote_lab/settings.py 中编辑或加入以下配置,避免保留同名配置的多个有效定义:
ROBOTSTXT_OBEY = TrueCONCURRENT_REQUESTS = 4CONCURRENT_REQUESTS_PER_DOMAIN = 2DOWNLOAD_DELAY = 1.0DOWNLOAD_TIMEOUT = 15AUTOTHROTTLE_ENABLED = TrueAUTOTHROTTLE_TARGET_CONCURRENCY = 1.0RETRY_TIMES = 2DEPTH_LIMIT = 3FEED_EXPORT_ENCODING = "utf-8"DEPTH_LIMIT=3 限制请求链深度,用来缩小练习范围;它不是精确的页面数量限制。DOWNLOAD_DELAY 与自动限速共同约束抓取,实际请求间隔会变化。重试、并发与延迟应按站点承载能力设置,不能把遭到限流当作继续增大并发的理由。Scrapy 配置参考
3.9.5 Pipeline:校验、清洗和去重
把下面的类加入 quote_lab/pipelines.py:
from scrapy.exceptions import DropItem
class CleanQuotePipeline: def __init__(self): self.seen = set()
def process_item(self, item, spider): text = item.get("text") author = item.get("author") if not isinstance(text, str) or not isinstance(author, str): raise DropItem("缺少正文或作者") text, author = text.strip(), author.strip() if not text or not author: raise DropItem("正文或作者为空") key = (text, author) if key in self.seen: raise DropItem("重复记录") self.seen.add(key) item["text"], item["author"] = key return item再在 settings.py 中注册:
ITEM_PIPELINES = {"quote_lab.pipelines.CleanQuotePipeline": 300}数字越小越早执行;管道应返回清洗后的条目或抛出 DropItem。这里的集合只在本次进程中去重,大规模或跨次运行去重应改用数据库唯一约束,避免内存不断增长。
3.9.6 运行、导出和定位问题
在有 scrapy.cfg 的项目根目录执行:
scrapy listscrapy crawl quotes -O quotes.jsonl-O 覆盖同名输出文件,重新运行前确认不需要保留旧结果;-o 对支持追加的输出格式通常追加,反复运行可能产生重复数据。这里选择 JSONL,便于逐条检查。
执行完后,用以下脚本检查前两条并统计数量:
import json
count = 0with open("quotes.jsonl", encoding="utf-8") as file: for line in file: item = json.loads(line) if not item.get("text") or not item.get("author"): raise ValueError("输出存在无效条目") count += 1 if count <= 2: print(item["author"], item["text"])print("有效条目数:", count)| 问题 | 排查方向 |
|---|---|
| 找不到 Spider | 执行目录、name、模块导入错误、当前环境 |
| 返回 200 却没有数据 | 选择器是否匹配实际 HTML,是否只有 JS 渲染后才有内容 |
| 403 / 429 或 robots 拒绝 | 权限、站点规则、访问频率;停止或调整合法访问方式 |
字段大量为 None | 检查页面结构,区分缺失字段与选择器错误 |
| 下一页没有抓取 | 相对链接、域名限制、深度限制、重复请求过滤 |
可以使用 scrapy shell https://quotes.toscrape.com/ 单独验证选择器,再运行整个爬虫。查看日志中的响应状态、item_scraped_count、丢弃原因和结束原因,不能只凭程序退出判断采集成功。
完成这个练习后,可将 JSONL 清洗结果接入 3.8 节的 SQLite,把唯一键设为稳定的业务标识,补充采集时间与来源,并对重跑、空页面和网络失败进行验证。
如果这篇文章对你有帮助,欢迎分享给更多人!
部分信息可能已经过时
