Python
python
环境搭建
pyenv
1 | # Python 版本管理工具，能让你在不同 Python 版本间轻松切换。可以结合 venv 等虚拟环境工具来管理虚拟环境。 |
venv
1 | # Python 标准库自带的虚拟环境创建工具，用于创建轻量级的虚拟环境。它只能管理虚拟环境，无法管理 Python 版本 |
uv
install
- linux / mac : curl -LsSf https://astral.sh/uv/install.sh | sh
创建项目
1 | > uv init tmp |
操作环境
同步项目依赖
1
2
3
4
5
6> uv sync
> uv sync --index-url 源
Using CPython 3.10.18 interpreter at: /Users/yzhyi/.pyenv/versions/3.10.18/bin/python3.10
Creating virtual environment at: .venv
Resolved 1 package in 3ms
Audited in 0.05ms同步之后，会自动查找或下载合适的 Python 版本，创建并设置项目的虚拟环境，构建完整的依赖列表并写入 uv.lock 文件，最后将依赖同步到虚拟环境中。
既然使用uv管理项目的话，就使用uv的命令来运行代码，不要像以前那样使用python xxx.py来运行。
1
2> uv run main.py
Hello from tmp!
管理依赖
1 | uv add pandas |
区分开发和生产环境
- 把pandas安装到开发环境中，而把requests安装到生产环境中。
1
2uv add --group dev pandas
uv add --group production requests
管理多版本 Python
1 | uv python install |
基础语法
基本数据类型
Number（数字）
- 数字是不允许改变的，这就意味着如果改变数字数据类型的值，将重新分配内存空间。
- 可以使用del语句删除一些数字对象的引用(del var)
String（字符串）
- 字符串用单引号 ‘ 或双引号 “ 括起来，同时使用反斜杠 \ 转义特殊字符。
- 字符串的截取的语法格式如下：变量[头下标:尾下标] (索引值以 0 为开始值，-1 为从末尾的开始位置。)
- 与 C 字符串不同的是，Python 字符串不能被改变。向一个索引位置赋值，比如 word[0] = ‘m’ 会导致错误。
- 字符串可以用+运算符连接在一起，用*运算符重复。
- 不支持单字符类型，单字符在 Python 中也是作为一个字符串使用。
- f-string 是 python3.6 之后版本添加的，称之为字面量格式化字符串，是新的格式化字符串的语法。f-string 格式化字符串以 f 开头，后面跟着字符串，字符串中的表达式用大括号 {} 包起来，它会将变量或表达式计算后的值替换进去。
1
2s = 'hello'
print(s[1:2]) # 输出 e (左闭右开)
bool（布尔类型）
- bool 是 int 的子类，因此布尔值可以被看作整数来使用，其中 True 等价于 1。
- 布尔类型可以和其他数据类型进行比较，比如数字、字符串等。在比较时，Python 会将 True 视为 1，False 视为 0。
- 可以使用 bool() 函数将其他类型的值转换为布尔值。以下值在转换为布尔值时为 False：None、False、零 (0、0.0、0j)、空序列（如 ‘’、()、[]）和空映射（如 {}）。其他所有值转换为布尔值时均为 True。
- 所有非零的数字和非空的字符串、列表、元组等数据类型都被视为 True，只有 0、空字符串、空列表、空元组等被视为 False。
List（列表）
- 列表截取的语法格式与 String 相同。(左闭右开)
- 列表中的元素是可以改变的
1
2
3a = [9, 2, 13, 14, 15, 6]
a[2:5] = [] # 将对应的元素值设置为 []
# a 变成 [9, 2, 6] - 列表可以使用 + 操作符进行拼接。
- Python 列表截取可以接收第三个参数，参数作用是截取的步长。(list[1:4:2], 2 是步长)
- 更新列表：append()
- 删除列表元素：del list[index]
Tuple（元组）
- 元组与列表类似，不同之处在于元组的元素不能修改。元组写在小括号 () 里，元素之间用逗号隔开。
- 虽然tuple的元素不可改变，但它可以包含可变的对象，比如list列表。
- 创建只有一个元素的元组，需要注意在元素后面添加一个逗号，以区分它是一个元组而不是一个普通的值。
- 元组也可以使用 + 操作符进行拼接。
Set（集合）
- Python 中的集合（Set）是一种无序、可变的数据类型，用于存储唯一的元素。
- 集合中的元素不会重复，并且可以进行交集、并集、差集等常见的集合操作。
- 在 Python 中，集合使用大括号 {} 表示，元素之间用逗号 , 分隔。
- 添加元素：s.add( x )
- 删除元素：s.remove( x )
- 计算元素个数：len(s)
- 清空集合：s.clear()
- 判断元素是否在集合中：x in s
Dictionary（字典）
- 字典是一种映射类型，字典用 { } 标识，它是一个无序的 键(key) : 值(value) 的集合。
- 键(key)必须使用不可变类型。
数据类型转换
- 不允许整型数据与字符串类型的数据进行相加
- 显式类型转换： int(), float(), str()
运算符
- /: 精确除法
- //: 向下取整除法(若分子或分母中有小数，则得到的是取整的小数)
- **: 幂运算
- 逻辑运算: and、or、not
- 成员运算符: in、not in
- 身份运算符: is、is not (判断两个标识符是不是引用自一个对象)
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#!/usr/bin/python3
a = 20
b = 20
if ( a is b ):
print ("1 - a 和 b 有相同的标识")
else:
print ("1 - a 和 b 没有相同的标识")
if ( id(a) == id(b) ):
print ("2 - a 和 b 有相同的标识")
else:
print ("2 - a 和 b 没有相同的标识")
# 修改变量 b 的值
b = 30
if ( a is b ):
print ("3 - a 和 b 有相同的标识")
else:
print ("3 - a 和 b 没有相同的标识")
if ( a is not b ):
print ("4 - a 和 b 没有相同的标识")
else:
print ("4 - a 和 b 有相同的标识")
条件控制
1 | if condition_1: |
循环语句
- while 循环 (没有 do while 循环)
1
2
3
4
5
6n = 100
sum = 0
counter = 1
while counter <= n:
sum = sum + counter
counter += 1 - for 循环
1
2for <variable> in <sequence>:
<statements>
推导式
列表推导式
1
2
3
4
5
6
7
8
9
10[表达式 for 变量 in 列表]
[out_exp_res for out_exp in input_list]
或者
[表达式 for 变量 in 列表 if 条件]
[out_exp_res for out_exp in input_list if condition]
names = ['Bob','Tom','alice','Jerry','Wendy','Smith']
new_names = [name.upper()for name in names if len(name)>3]
print(new_names)
['ALICE', 'JERRY', 'WENDY', 'SMITH']字典推导式
1
2
3
4
5
6
7
8
9{ key_expr: value_expr for value in collection }
或
{ key_expr: value_expr for value in collection if condition }
listdemo = ['Google','Runoob', 'Taobao']
# 将列表中各字符串值为键，各字符串的长度为值，组成键值对
newdict = {key:len(key) for key in listdemo}
newdict
{'Google': 6, 'Runoob': 6, 'Taobao': 6}集合推导式
1
2
3
4
5
6
7{ expression for item in Sequence }
或
{ expression for item in Sequence if conditional }
setnew = {i**2 for i in (1,2,3)}
setnew
{1, 4, 9}元组推导式（生成器表达式）
1
2
3
4
5
6
7
8
9
10(expression for item in Sequence )
或
(expression for item in Sequence if conditional )
# 元组推导式和列表推导式的用法也完全相同，只是元组推导式是用 () 圆括号将各部分括起来，而列表推导式用的是中括号 []，另外元组推导式返回的结果是一个生成器对象。
a = (x for x in range(1,10))
a
<generator object <genexpr> at 0x7faf6ee20a50> # 返回的是生成器对象
tuple(a) # 使用 tuple() 函数，可以直接将生成器对象转换成元组
(1, 2, 3, 4, 5, 6, 7, 8, 9)
迭代器与生成器
迭代器
迭代器是一个可以记住遍历的位置的对象。
迭代器对象从集合的第一个元素开始访问，直到所有的元素被访问完结束。迭代器只能往前不会后退。
迭代器有两个基本的方法：iter() 和 next()。
字符串，列表或元组对象都可用于创建迭代器：
1
2
3
4
5
6
7list=[1,2,3,4]
it = iter(list) # 创建迭代器对象
print (next(it)) # 输出迭代器的下一个元素
1
print (next(it))
2
>>>1
2
3
4
5# 迭代器对象可以使用常规for语句进行遍历：
list=[1,2,3,4]
it = iter(list) # 创建迭代器对象
for x in it:
print (x, end=" ")1
2
3
4
5
6
7
8
9
10
11# 也可以使用 next() 函数：
import sys # 引入 sys 模块
list=[1,2,3,4]
it = iter(list) # 创建迭代器对象
while True:
try:
print (next(it))
except StopIteration:
sys.exit()创建一个迭代器: 把一个类作为一个迭代器使用需要在类中实现两个方法 iter() 与 next() 。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18class MyNumbers:
def __iter__(self):
self.a = 1
return self
def __next__(self):
x = self.a
self.a += 1
return x
myclass = MyNumbers()
myiter = iter(myclass)
print(next(myiter))
print(next(myiter))
print(next(myiter))
print(next(myiter))
print(next(myiter))
生成器
- 使用了 yield 的函数被称为生成器（generator）。
- yield 是一个关键字，用于定义生成器函数，生成器函数是一种特殊的函数，可以在迭代过程中逐步产生值，而不是一次性返回所有结果。
- 跟普通函数不同的是，生成器是一个返回迭代器的函数，只能用于迭代操作，更简单点理解生成器就是一个迭代器。
- 当在生成器函数中使用 yield 语句时，函数的执行将会暂停，并将 yield 后面的表达式作为当前迭代的值返回。
- 每次调用生成器的 next() 方法或使用 for 循环进行迭代时，函数会从上次暂停的地方继续执行，直到再次遇到 yield 语句。这样，生成器函数可以逐步产生值，而不需要一次性计算并返回所有结果。
- 调用一个生成器函数，返回的是一个迭代器对象。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16def countdown(n):
while n > 0:
yield n
n -= 1
# 创建生成器对象
generator = countdown(5)
# 通过迭代生成器获取值
print(next(generator)) # 输出: 5
print(next(generator)) # 输出: 4
print(next(generator)) # 输出: 3
# 使用 for 循环迭代生成器
for value in generator:
print(value) # 输出: 2 1
with 关键字
with 是 Python 中的一个关键字，用于上下文管理协议（Context Management Protocol）。它简化了资源管理代码，特别是那些需要明确释放或清理的资源（如文件、网络连接、数据库连接等）。
优势：
- 自动资源释放：确保资源在使用后被正确关闭
- 代码简洁：减少样板代码
- 异常安全：即使在代码块中发生异常，资源也会被正确释放
- 可读性强：明确标识资源的作用域
基础用法
1
2
3
4
5with expression [as variable]:
# 代码块
# expression 返回一个支持上下文管理协议的对象
# as variable 是可选的，用于将表达式结果赋值给变量
# 代码块执行完毕后，自动调用清理方法工作原理
- with 语句背后是 Python 的上下文管理协议，该协议要求对象实现两个方法：
- enter()：进入上下文时调用，返回值赋给 as 后的变量
- exit()：退出上下文时调用，处理清理工作
- with 语句背后是 Python 的上下文管理协议，该协议要求对象实现两个方法：
创建自定义的上下文管理器
- 类实现方式, 可以通过实现 enter 和 exit 方法创建自定义的上下文管理器：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18class Timer:
def __enter__(self):
import time
self.start = time.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
import time
self.end = time.time()
print(f"耗时: {self.end - self.start:.2f}秒")
return False
# 使用示例
with Timer() as t:
# 执行一些耗时操作
sum(range(1000000))
a = 1
print(a) # 正确，with 与 if 这种语句块不同，并不会限制变量的作用域（a 的作用于未被限制在 with 中，而是在外层） - 使用 contextlib 模块, Python 的 contextlib 模块提供了更简单的方式来创建上下文管理器：
1
2
3
4
5
6
7
8
9
10
11from contextlib import contextmanager
def tag(name):
print(f"<{name}>")
yield
print(f"</{name}>")
# 使用示例
with tag("h1"):
print("这是一个标题")
- 类实现方式, 可以通过实现 enter 和 exit 方法创建自定义的上下文管理器：
函数
定义函数使用 def 关键字，一般格式如下：
1
2def 函数名（参数列表）:
函数体参数传递
可更改(mutable)与不可更改(immutable)对象: 在 python 中，strings, tuples, 和 numbers 是不可更改的对象，而 list,dict 等则是可以修改的对象。
- 不可变类型：变量赋值 a=5 后再赋值 a=10，这里实际是新生成一个 int 值对象 10，再让 a 指向它，而 5 被丢弃，不是改变 a 的值，相当于新生成了 a。
- 可变类型：变量赋值 la=[1,2,3,4] 后再赋值 la[2]=5 则是将 list la 的第三个元素值更改，本身la没有动，只是其内部的一部分值被修改了。
python 函数的参数传递：
- 不可变类型：类似 C++ 的值传递，如整数、字符串、元组。如 fun(a)，传递的只是 a 的值，没有影响 a 对象本身。如果在 fun(a) 内部修改 a 的值，则是新生成一个 a 的对象。
- 可变类型：类似 C++ 的引用传递，如 列表，字典。如 fun(la)，则是将 la 真正的传过去，修改后 fun 外部的 la 也会受影响
不定长参数
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
32
33
34
35
36
37def functionname([formal_args,] *var_args_tuple ):
"函数_文档字符串"
function_suite
return [expression]
# 举例
def printinfo( arg1, *vartuple ):
"打印任何传入的参数"
print ("输出: ")
print (arg1)
print (vartuple)
# 调用printinfo 函数
printinfo( 70, 60, 50 )
# 输出:
70
(60, 50)
# 还有一种就是参数带两个星号 **基本语法如下：
def functionname([formal_args,] **var_args_dict ):
"函数_文档字符串"
function_suite
return [expression]
# 加了两个星号 ** 的参数会以字典的形式导入。
# 举例
def printinfo( arg1, **vardict ):
"打印任何传入的参数"
print ("输出: ")
print (arg1)
print (vardict)
# 调用printinfo 函数
printinfo(1, a=2,b=3)
# 输出
1
{'a': 2, 'b': 3}
lambda 函数
- ambda 函数是一种小型、匿名的、内联函数，它可以具有任意数量的参数，但只能有一个表达式。
- lambda 语法格式：
1
2
3
4
5lambda arguments: expression
lambda是 Python 的关键字，用于定义 lambda 函数。
arguments 是参数列表，可以包含零个或多个参数，但必须在冒号(:)前指定。
expression 是一个表达式，用于计算并返回函数的结果。
装饰器
- 装饰器（decorators）是 Python 中的一种高级功能，它允许你动态地修改函数或类的行为。装饰器是一种函数，它接受一个函数作为参数，并返回一个新的函数或修改原来的函数。
- 装饰器的语法使用 @decorator_name 来应用在函数或方法上。
函数装饰器
- Python 装饰允许在不修改原有函数代码的基础上，动态地增加或修改函数的功能，装饰器本质上是一个接收函数作为输入并返回一个新的包装过后的函数的对象。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17def decorator_function(original_function):
def wrapper(*args, **kwargs):
# 这里是在调用原始函数前添加的新功能
before_call_code()
result = original_function(*args, **kwargs)
# 这里是在调用原始函数后添加的新功能
after_call_code()
return result
return wrapper
# 使用装饰器
def target_function(arg1, arg2):
pass # 原始函数的实现
类装饰器
- 类装饰器（Class Decorator）是一种用于动态修改类行为的装饰器，它接收一个类作为参数，并返回一个新的类或修改后的类。
- 函数形式的类装饰器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24def log_class(cls):
"""类装饰器，在调用方法前后打印日志"""
class Wrapper:
def __init__(self, *args, **kwargs):
self.wrapped = cls(*args, **kwargs) # 实例化原始类
def __getattr__(self, name):
"""拦截未定义的属性访问，转发给原始类"""
return getattr(self.wrapped, name)
def display(self):
print(f"调用 {cls.__name__}.display() 前")
self.wrapped.display()
print(f"调用 {cls.__name__}.display() 后")
return Wrapper # 返回包装后的类
class MyClass:
def display(self):
print("这是 MyClass 的 display 方法")
obj = MyClass()
obj.display() - 类形式的类装饰器（实现 call 方法）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20class SingletonDecorator:
"""类装饰器，使目标类变成单例模式"""
def __init__(self, cls):
self.cls = cls
self.instance = None
def __call__(self, *args, **kwargs):
"""拦截实例化过程，确保只创建一个实例"""
if self.instance is None:
self.instance = self.cls(*args, **kwargs)
return self.instance
class Database:
def __init__(self):
print("Database 初始化")
db1 = Database()
db2 = Database()
print(db1 is db2) # True，说明是同一个实例
内置装饰器
- @staticmethod: 将方法定义为静态方法，不需要实例化类即可调用。
- @classmethod: 将方法定义为类方法，第一个参数是类本身（通常命名为 cls）。
- @property: 将方法转换为属性，使其可以像属性一样访问。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24class MyClass:
def static_method():
print("This is a static method.")
def class_method(cls):
print(f"This is a class method of {cls.__name__}.")
def name(self):
return self._name
def name(self, value):
self._name = value
# 使用
MyClass.static_method()
MyClass.class_method()
obj = MyClass()
obj.name = "Alice"
print(obj.name)
模块
- 一个模块只会被导入一次，不管你执行了多少次 import。这样可以防止导入模块被一遍又一遍地执行。
- 搜索路径
- 当前目录。
- 环境变量 PYTHONPATH 指定的目录。
- Python 标准库目录。
- .pth 文件中指定的目录。
- import 模块名：并没有把直接定义在模块中的函数名称写入到当前符号表里，只是把模块的名字写到了那里, 可以使用模块名称来访问函数.
- from … import 语句: 从模块中导入一个指定的部分到当前命名空间中
- 给模块起别名: 使用 as 关键字为模块或函数起别名
- 模块除了方法定义，还可以包括可执行的代码。这些代码一般用来初始化这个模块。这些代码只有在第一次被导入时才会被执行。
- 每个模块有各自独立的符号表，在模块内部为所有的函数当作全局符号表来使用。
- 包
- 包是一种管理 Python 模块命名空间的形式，采用”点模块名称”。
- 目录只有包含一个叫做 init.py 的文件才会被认作是一个包，主要是为了避免一些滥俗的名字（比如叫做 string）不小心的影响搜索路径中的有效模块。
- 最简单的情况，放一个空的 :file:init.py就可以了。当然这个文件中也可以包含一些初始化代码.
错误和异常
1 | try: |
- 自定义异常
- 异常类继承自 Exception 类，可以直接继承，或者间接继承
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16class MyError(Exception):
def __init__(self, value): # 类 Exception 默认的 __init__() 被覆盖。
self.value = value
def __str__(self):
return repr(self.value)
try:
raise MyError(2*2)
except MyError as e:
print('My exception occurred, value:', e.value)
My exception occurred, value: 4
raise MyError('oops!')
Traceback (most recent call last):
File "<stdin>", line 1, in ?
__main__.MyError: 'oops!'
- 异常类继承自 Exception 类，可以直接继承，或者间接继承
面向对象
- 类有一个名为 init() 的特殊方法（构造方法），该方法在类实例化时会自动调用
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#类定义
class people:
#定义基本属性
name = ''
age = 0
#定义私有属性,私有属性在类外部无法直接进行访问
__weight = 0
#定义构造方法
def __init__(self,n,a,w):
self.name = n
self.age = a
self.__weight = w
def speak(self):
print("%s 说: 我 %d 岁。" %(self.name,self.age))
#单继承示例
class student(people):
grade = ''
def __init__(self,n,a,w,g):
#调用父类的构函
people.__init__(self,n,a,w)
self.grade = g
#覆写父类的方法
def speak(self):
print("%s 说: 我 %d 岁了，我在读 %d 年级"%(self.name,self.age,self.grade))
s = student('ken',10,60,3)
s.speak() - 运算符重载
1
2
3
4
5
6
7
8
9
10
11
12
13
14class Vector:
def __init__(self, a, b):
self.a = a
self.b = b
def __str__(self):
return 'Vector (%d, %d)' % (self.a, self.b)
def __add__(self,other):
return Vector(self.a + other.a, self.b + other.b)
v1 = Vector(2,10)
v2 = Vector(5,-2)
print (v1 + v2)
命名空间和作用域
- 当内部作用域想修改外部作用域的变量时，就要用到 global 和 nonlocal 关键字了。
1
2
3
4
5
6
7
8num = 1
def fun1():
global num # 需要使用 global 关键字声明
print(num)
num = 123
print(num)
fun1()
print(num)
多线程
- threading 模块
- threading.current_thread(): 返回当前的线程变量。
- threading.enumerate(): 返回一个包含正在运行的线程的列表。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
- threading.active_count(): 返回正在运行的线程数量，与 len(threading.enumerate()) 有相同的结果。
- threading.Thread(target, args=(), kwargs={}, daemon=None)：创建Thread类的实例。
- target：线程将要执行的目标函数。
- args：目标函数的参数，以元组形式传递。
- kwargs：目标函数的关键字参数，以字典形式传递。
- daemon：指定线程是否为守护线程。
- threading.Thread 类提供了以下方法与属性:
- init(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)：初始化Thread对象。
- group：线程组，暂时未使用，保留为将来的扩展。
- target：线程将要执行的目标函数。
- name：线程的名称。
- args：目标函数的参数，以元组形式传递。
- kwargs：目标函数的关键字参数，以字典形式传递。
- daemon：指定线程是否为守护线程。
- start(self)：启动线程。将调用线程的run()方法。
- join(self, timeout=None)：等待线程终止。默认情况下，join()会一直阻塞，直到被调用线程终止。如果指定了timeout参数，则最多等待timeout秒。
- is_alive(self)：返回线程是否在运行。如果线程已经启动且尚未终止，则返回True，否则返回False。
1
2
3
4
5
6
7
8
9
10
11
12# Code to execute in an independent thread
import time
def countdown(n):
while n > 0:
print('T-minus', n)
n -= 1
time.sleep(5)
# Create and launch a thread
from threading import Thread
t = Thread(target=countdown, args=(10,))
t.start()
- init(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)：初始化Thread对象。
- 线程同步:
- 使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法
- threadLock = threading.Lock()
- 使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法
json 数据解析
- json.dumps(): 对数据进行编码。
1
2
3
4
5
6
7
8
9
10
11import json
# Python 字典类型转换为 JSON 对象
data = {
'no' : 1,
'name' : 'Runoob',
'url' : 'https://www.runoob.com'
}
json_str = json.dumps(data)
print ("Python 原始数据：", repr(data))
print ("JSON 对象：", json_str) - json.loads(): 对数据进行解码。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import json
# Python 字典类型转换为 JSON 对象
data1 = {
'no' : 1,
'name' : 'Runoob',
'url' : 'http://www.runoob.com'
}
json_str = json.dumps(data1)
print ("Python 原始数据：", repr(data1))
print ("JSON 对象：", json_str)
# 将 JSON 对象转换为 Python 字典
data2 = json.loads(json_str)
print ("data2['name']: ", data2['name'])
print ("data2['url']: ", data2['url']) - 处理的是文件而不是字符串，你可以使用 json.dump() 和 json.load() 来编码和解码JSON数据。
1
2
3
4
5
6
7# 写入 JSON 数据
with open('data.json', 'w') as f:
json.dump(data, f)
# 读取数据
with open('data.json', 'r') as f:
data = json.load(f) - Python 编码为 JSON 类型转换对应表
- dict object
- list, tuple array
- str string
- int, float, int- & float-derived Enums number
- True true
- False false
- None null
- JSON 解码为 Python 类型转换对应表：
- object dict
- array list
- string str
- number (int) int
- number (real) float
- true True
- false False
- null None
requests 模块
- Python requests 是一个常用的 HTTP 请求库，可以方便地向网站发送 HTTP 请求，并获取响应结果。
1
2
3
4
5
6
7
8
9
10import requests
# 发送请求
x = requests.get('https://www.runoob.com/')
# 返回网页内容
print(x.text)
# 每次调用 requests 请求之后，会返回一个 response 对象，该对象包含了具体的响应信息，如状态码、响应头、响应内容等：
print(response.status_code) # 获取响应状态码
print(response.headers) # 获取响应头
print(response.content) # 获取响应内容
queue 模块
- queue 模块提供了一个线程安全的队列实现，用于在多线程编程中安全地传递数据。
队列是一种先进先出（FIFO）的数据结构，queue 模块提供了多种队列类型，包括 Queue、LifoQueue 和 PriorityQueue，以满足不同的需求。
Queue
1 | import queue |
LifoQueue
- LifoQueue 是一种后进先出（LIFO）的队列，类似于栈。
1
2
3
4
5
6
7
8
9
10
11
12
13
14import queue
# 创建一个 LIFO 队列
q = queue.LifoQueue()
# 向队列中添加元素
q.put(1)
q.put(2)
q.put(3)
# 从队列中获取元素
print(q.get()) # 输出: 3
print(q.get()) # 输出: 2
print(q.get()) # 输出: 1
PriorityQueue
- PriorityQueue 是一种优先级队列，元素按照优先级顺序被取出。
1
2
3
4
5
6
7
8
9
10
11
12
13
14import queue
# 创建一个优先级队列
q = queue.PriorityQueue()
# 向队列中添加元素，元素为元组 (优先级, 数据)
q.put((3, 'Low priority'))
q.put((1, 'High priority'))
q.put((2, 'Medium priority'))
# 从队列中获取元素
print(q.get()) # 输出: (1, 'High priority')
print(q.get()) # 输出: (2, 'Medium priority')
print(q.get()) # 输出: (3, 'Low priority')
logging 模块
配置日志级别
- DEBUG：详细的调试信息，通常用于开发阶段。
- INFO：程序正常运行时的信息。
- WARNING：表示潜在的问题，但程序仍能正常运行。
- ERROR：表示程序中的错误，导致某些功能无法正常工作。
- CRITICAL：表示严重的错误，可能导致程序崩溃。
1
logging.basicConfig(level=logging.DEBUG)
记录日志
1 | logging.debug("这是一条调试信息") |
日志输出格式
- 通过 basicConfig 方法自定义日志的输出格式。
1
2
3
4
5logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
将日志输出到文件
1 | logging.basicConfig( |
高级用法
使用多个日志记录器
- 在大型项目中，你可能需要为不同的模块或组件创建独立的日志记录器。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25import logging
logger = logging.getLogger("my_logger")
logger.setLevel(logging.DEBUG)
# 创建文件处理器
file_handler = logging.FileHandler("my_logger.log")
file_handler.setLevel(logging.DEBUG)
# 创建控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 设置日志格式
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# 将处理器添加到日志记录器
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# 记录日志
logger.debug("这是一条调试信息") # file
logger.info("这是一条普通信息") # file and console
日志过滤器
- 通过过滤器来控制哪些日志需要被记录。
1
2
3
4
5class MyFilter(logging.Filter):
def filter(self, record):
return record.levelno == logging.ERROR
logger.addFilter(MyFilter())
日志轮转
- 当日志文件过大时，可以使用 RotatingFileHandler 或 TimedRotatingFileHandler 实现日志轮转
1
2
3
4from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler("app.log", maxBytes=1024, backupCount=3)
logger.addHandler(handler)
并发编程
多线程编程
- 使用 threading 模块可以创建多个线程来并发执行任务。然而，Python 的全局解释锁（GIL）可能会限制多线程的并行性，尤其在 CPU 密集型任务上效果有限。
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
32
33
34import concurrent.futures
import threading
import time
# 定义线程要执行的函数
def worker(index):
print(f"Thread {index} is starting")
time.sleep(2)
print(f"Thread {index} is done")
start = time.time()
# 使用线程池创建多线程
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = [executor.submit(worker, i) for i in range(5)]
# 等待所有线程完成
concurrent.futures.wait(futures)
end = time.time()
print("concurrent cost: ", end - start, " seconds")
print("All threads are done")
### print
Thread 0 is starting
Thread 1 is starting
Thread 2 is starting
Thread 0 is done
Thread 3 is starting
Thread 1 is done
Thread 2 is done
Thread 4 is starting
Thread 3 is done
Thread 4 is done
# 有任务执行完才会轮到下一个任务
多进程编程
- 使用 multiprocessing 模块可以创建多个进程来并发执行任务。每个进程都有自己的解释器和独立的内存空间，可以有效地利用多核处理器，适用于 CPU 密集型任务。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19import concurrent.futures
import time
# 定义进程要执行的函数
def worker(index):
print(f"Process {index} is starting")
# 模拟耗时操作
time.sleep(2)
print(f"Process {index} is done")
# 使用进程池创建多进程
# max_workers=3 表示创建一个最大容量为 3 的进程池，即同时可以运行的进程数量不会超过 3 个。
with concurrent.futures.ProcessPoolExecutor(max_workers=3) as executor:
# 提交多个进程任务 创建一个包含 5 个任务的列表，每个任务都会执行 worker 函数，传递不同的索引值作为参数。
futures = [executor.submit(worker, i) for i in range(5)]
# 等待所有进程完成
concurrent.futures.wait(futures)
print("All processes are done")
异步编程
全局解释器锁（GIL）
- 计算机程序设计语言解释器用于同步线程的一种机制，它使得任何时刻只有一个线程在执行；即使在多核处理器上，使用GIL 的解释器也只允许同一时间执行一个线程；所以当GIL存在的时候，即使电脑有多核cpu，单个时刻也只能够执行1个，相比并发加速的c++/java 所以慢；
- 多线程 threading 机制依然是有用的，用于IO 密集型计算，因为在IO期间，线程会释放GIL，实现CPU 和IO的并行，因此多线程用于IO密集型计算依然可以大幅度提升速度，但是多线程用于CPU密集型计算的时候，只会更加拖慢速度；使用multiprocessing 的多进程机制实现并行计算，利用多核cpu 的优势，为了应对GIL 的问题，python 提供了multiprocessing。
asyncio 模块
- 协程是 asyncio 的核心概念之一。它是一个特殊的函数，可以在执行过程中暂停，并在稍后恢复执行。协程通过 async def 关键字定义，并通过 await 关键字暂停执行，等待异步操作完成。
- 事件循环是 asyncio 的核心组件，负责调度和执行协程。它不断地检查是否有任务需要执行，并在任务完成后调用相应的回调函数。
- 任务是对协程的封装，表示一个正在执行或将要执行的协程。你可以通过 asyncio.create_task() 函数创建任务，并将其添加到事件循环中。
1
2
3async def main():
task = asyncio.create_task(say_hello())
await task
基本用法
运行协程
- 要运行一个协程，你可以使用 asyncio.run() 函数。它会创建一个事件循环，并运行指定的协程。
1
2
3
4
5
6
7
8import asyncio
async def main():
print("Start")
await asyncio.sleep(1)
print("End")
asyncio.run(main())
并发执行多个任务
- 使用 asyncio.gather() 函数并发执行多个协程，并等待它们全部完成。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import asyncio
async def task1():
print("Task 1 started")
await asyncio.sleep(1)
print("Task 1 finished")
async def task2():
print("Task 2 started")
await asyncio.sleep(2)
print("Task 2 finished")
async def main():
await asyncio.gather(task1(), task2())
asyncio.run(main())
超时控制
- 使用 asyncio.wait_for() 函数为协程设置超时时间。如果协程在指定时间内未完成，将引发 asyncio.TimeoutError 异常。
1
2
3
4
5
6
7
8
9
10
11
12
13import asyncio
async def long_task():
await asyncio.sleep(10)
print("Task finished")
async def main():
try:
await asyncio.wait_for(long_task(), timeout=5)
except asyncio.TimeoutError:
print("Task timed out")
asyncio.run(main())
logging
- 默认只会打印 warning 以上的信息
- logging.getLogger() 的基本功能是获取一个指定名称的日志记录器对象。
- logging.getLogger()（无参数）：获取一个根日志记录器（root logger）对象。默认的根日志记录器名称为空字符串 “”。
- logging.getLogger(name)（有参数）：获取一个指定名称的日志记录器对象。如果指定的名称在当前日志层次结构中不存在，它会创建一个新日志记录器。
- 每个获取的 logging 是不同的
pydantic
BaseModel
- BaseModel 是 Pydantic 库的核心基类，用于创建具有数据验证、序列化和类型检查功能的数据模型。它是现代 Python 开发中处理数据的强大工具。
数据验证和类型转换
1 | from pydantic import BaseModel, ValidationError, Field |
序列化和反序列化
1 | class Order(BaseModel): |
Field
1 | from pydantic import BaseModel, Field |
CamelModel
1 | from pydantic import BaseModel |