装饰器是Python中一种强大的语法糖，它可以在不修改原函数的情况下动态地扩展函数的功能。装饰器本质上是一个接受函数作为参数并返回新函数的高阶函数。
装饰器的核心思想是将函数作为参数传入另一个函数，在原函数执行前后添加额外的功能。这类似于面向切面编程（AOP）的概念，让我们能够在不修改原函数代码的情况下增强其行为。
# 装饰器的基本结构
def my_decorator(func):
def wrapper(*args, **kwargs):
# 在函数执行前的操作
print("函数执行前...")
# 执行原函数
result = func(*args, **kwargs)
# 在函数执行后的操作
print("函数执行后...")
return result
return wrapper
@my_decorator
def say_hello():
print("Hello!")
# 调用装饰后的函数
say_hello()
# 输出：
# 函数执行前...
# Hello!
# 函数执行后...
在上述代码中，@my_decorator是装饰器语法糖，等价于say_hello = my_decorator(say_hello)。装饰器将原始的say_hello函数替换为wrapper函数，从而在保持原函数功能的基础上添加了额外的行为。
装饰器可以接受参数来实现更灵活的功能。这需要再包装一层函数来接收参数。
def repeat(times):
"""重复执行函数的装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
for _ in range(times):
result = func(*args, **kwargs)
return result
return wrapper
return decorator
@repeat(times=3)
def greet(name):
print(f"Hello, {name}!")
greet("张三")
# 输出：Hello, 张三! 三次
这个装饰器工厂模式允许我们在使用装饰器时传递参数。@repeat(times=3)首先调用repeat(3)返回一个装饰器，然后再装饰目标函数。
使用装饰器后，原函数的元信息（如__name__、__doc__等）会被覆盖为包装函数的信息。可以使用functools.wraps来保留原函数的信息。
import functools
def my_decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
"""这是包装函数"""
print("执行前")
result = func(*args, **kwargs)
print("执行后")
return result
return wrapper
@my_decorator
def example():
"""这是原函数的文档字符串"""
print("原函数执行")
print(example.__name__) # example（保留原函数名）
print(example.__doc__) # 这是原函数的文档字符串（保留原文档）
functools.wraps装饰器会自动复制被装饰函数的__name__、__doc__、__module__、__annotations__等属性到包装函数中，确保装饰后的函数保持与原函数相同的元信息。
装饰器不仅可用于函数，还可用于类。类装饰器可以修改类的行为或为类添加功能。
def singleton(cls):
"""单例模式装饰器"""
instances = {}
@functools.wraps(cls)
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
class Database:
def __init__(self):
print("创建数据库连接")
def query(self, sql):
return f"执行: {sql}"
# 测试
db1 = Database()
db2 = Database()
print(db1 is db2) # True（同一个实例）
这个单例装饰器确保一个类只有一个实例，多次实例化返回同一个对象，这在管理共享资源时非常有用。
Python提供了许多内置装饰器，了解它们有助于编写更好的代码。
| 装饰器 | 所属模块 | 说明 |
|---|---|---|
| @staticmethod | builtins | 静态方法装饰器 |
| @classmethod | builtins | 类方法装饰器 |
| @property | builtins | 属性装饰器 |
| @functools.wraps | functools | 保留函数元信息 |
| @functools.lru_cache | functools | 缓存装饰器 |
| @abstractmethod | abc | 抽象方法装饰器 |
class Circle:
def __init__(self, radius):
self.radius = radius
@property
def diameter(self):
return self.radius * 2
@property
def area(self):
import math
return math.pi * self.radius ** 2
circle = Circle(5)
print(circle.diameter) # 10（作为属性访问）
print(circle.area) # 78.54（作为属性访问）
@property装饰器允许我们将方法作为属性访问，提供了更直观的API，同时可以在访问器方法中添加计算逻辑。
迭代器是Python中用于遍历数据集合的对象，它必须实现__iter__和__next__方法。迭代器协议定义了对象如何被遍历的标准化方式。
__iter__方法返回迭代器对象本身，而__next__方法返回序列中的下一个元素，当没有更多元素时抛出StopIteration异常。
class MyIterator:
def __init__(self, data):
self.data = data
self.index = 0
def __iter__(self):
return self
def __next__(self):
if self.index >= len(self.data):
raise StopIteration
result = self.data[self.index]
self.index += 1
return result
# 使用迭代器
my_iter = MyIterator([1, 2, 3, 4, 5])
for item in my_iter:
print(item)
这个自定义迭代器实现了完整的迭代器协议，可以用在for循环中遍历数据。每次调用__next__都会返回下一个元素，直到数据遍历完毕。
生成器是一种特殊的迭代器，使用yield关键字来产生值，而不是return。生成器函数在每次调用时执行到yield语句，然后暂停执行，保存当前状态，下次调用时从暂停处继续执行。
生成器提供了一种简洁的方式来创建迭代器，特别适合处理大量数据或无限序列，因为生成器不会一次性将所有数据加载到内存中。
def fibonacci(n):
"""生成斐波那契数列"""
a, b = 0, 1
count = 0
while count < n:
yield a
a, b = b, a + b
count += 1
# 使用生成器
for num in fibonacci(10):
print(num, end=" ")
# 输出：0 1 1 2 3 5 8 13 21 34
生成器函数会在每次yield时暂停并返回当前值，下次调用时从上次暂停的地方继续执行。这使得生成器非常适合处理大型数据或无限序列，因为我们不需要一次性生成所有数据。
生成器表达式类似于列表推导式，但使用圆括号而不是方括号，返回一个生成器对象而不是列表。生成器表达式是创建迭代器的一种简洁方式。
# 列表推导式（立即生成所有元素）
squares_list = [x**2 for x in range(10)]
print(squares_list) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
# 生成器表达式（延迟生成，按需计算）
squares_gen = (x**2 for x in range(10))
print(squares_gen) # <generator object <genexpr> at 0x...>
# 使用next()获取元素
print(next(squares_gen)) # 0
print(next(squares_gen)) # 1
# 在for循环中使用
for sq in squares_gen:
print(sq)
生成器表达式的优势在于它不会立即计算所有值，而是按需生成，这在处理大型数据集时可以节省大量内存。生成器是惰性求值的，只有当需要时才计算下一个值。
itertools模块提供了许多用于操作迭代器的函数，是处理序列数据的有力工具。
| 函数 | 说明 |
|---|---|
| count(start, step) | 无限计数迭代器 |
| cycle(iterable) | 无限循环迭代器 |
| repeat(elem, n) | 重复元素n次 |
| chain(*iterables) | 连接多个迭代器 |
| islice(iterable, start, stop, step) | 切片迭代器 |
| tee(iterable, n) | 将一个迭代器复制n份 |
| filterfalse(pred, iterable) | 过滤不满足条件的元素 |
| accumulate(iterable, func) | 累加运算 |
import itertools
# count - 无限计数
for i in itertools.count(0, 2):
if i > 10:
break
print(i, end=" ") # 0 2 4 6 8 10
# chain - 连接多个序列
combined = itertools.chain([1, 2], [3, 4], [5])
print(list(combined)) # [1, 2, 3, 4, 5]
# islice - 切片
sliced = itertools.islice(range(10), 2, 8, 2)
print(list(sliced)) # [2, 4, 6]
# tee - 复制迭代器
iter1, iter2 = itertools.tee([1, 2, 3, 4, 5], 2)
print(list(iter1)) # [1, 2, 3, 4, 5]
print(list(iter2)) # [1, 2, 3, 4, 5]
上下文管理器是一种用于管理资源的协议，确保资源在使用后被正确释放。最常见的应用场景是文件操作，确保文件在使用后被关闭，即使发生异常也能正确关闭。
上下文管理器通过__enter__和__exit__方法实现。__enter__方法在进入with语句时调用，__exit__方法在离开with语句时调用。
class FileManager:
def __init__(self, filename, mode):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
# 返回True可以抑制异常传播
return False
# 使用上下文管理器
with FileManager("test.txt", "w") as f:
f.write("Hello, World!")
# 文件自动关闭
在这个例子中，__enter__方法打开文件并返回文件对象，__exit__方法确保文件被正确关闭。无论with块内是否发生异常，__exit__都会被调用，从而保证资源被正确释放。
Python的contextlib模块提供了contextmanager装饰器，可以用更简洁的方式创建上下文管理器，而不需要定义完整的类。
from contextlib import contextmanager
@contextmanager
def managed_resource(name):
"""资源管理上下文管理器"""
print(f"获取资源: {name}")
resource = f"资源-{name}"
try:
yield resource # 类似于return，将资源提供给with块
finally:
print(f"释放资源: {name}")
# 使用
with managed_resource("数据库") as res:
print(f"使用{res}进行操作")
# 输出：
# 获取资源: 数据库
# 使用资源-数据库进行操作
# 释放资源: 数据库
@contextmanager装饰器允许我们使用yield语句来分隔__enter__（yield之前）和__exit__（yield之后）的代码。这大大简化了上下文管理器的编写，特别适合简单的资源管理场景。
可以使用多个嵌套的with语句或在一个with语句中使用多个上下文管理器。
# 嵌套with语句
with open("file1.txt", "w") as f1:
with open("file2.txt", "w") as f2:
f1.write("内容1")
f2.write("内容2")
# 逗号分隔的多个上下文管理器
with open("file1.txt", "w") as f1, open("file2.txt", "w") as f2:
f1.write("内容1")
f2.write("内容2")
# Python 3.10+的括号语法
with (open("file1.txt", "w") as f1,
open("file2.txt", "w") as f2):
f1.write("内容1")
f2.write("内容2")
上下文管理器不仅用于文件操作，还广泛应用于需要资源管理或状态恢复的场景。
| 场景 | 说明 |
|---|---|
| 文件操作 | 自动关闭文件 |
| 数据库连接 | 自动提交或回滚事务 |
| 线程锁 | 自动获取和释放锁 |
| 临时状态 | 保存和恢复状态 |
| 计时 | 测量代码执行时间 |
| 模拟补丁 | 在测试中临时替换对象 |
import time
from contextlib import contextmanager
@contextmanager
def timer(name):
"""代码块计时器"""
start = time.time()
yield
end = time.time()
print(f"{name} 耗时: {end - start:.4f}秒")
# 使用
with timer("数据处理"):
time.sleep(0.1)
# 输出：数据处理 耗时: 0.1xxx秒
反射是指在运行时动态获取对象的类型、属性和方法的能力。Python提供了丰富的内置函数和模块来支持反射操作。
| 函数 | 说明 |
|---|---|
| type(obj) | 获取对象的类型 |
| isinstance(obj, class) | 判断对象是否是类的实例 |
| hasattr(obj, name) | 判断对象是否有指定属性 |
| getattr(obj, name, default) | 获取对象的属性值 |
| setattr(obj, name, value) | 设置对象的属性值 |
| delattr(obj, name) | 删除对象的属性 |
| callable(obj) | 判断对象是否可调用 |
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def greet(self):
return f"Hello, I'm {self.name}"
person = Person("张三", 25)
# 反射操作
print(type(person)) # <class '__main__.Person'>
print(isinstance(person, Person)) # True
print(hasattr(person, "name")) # True
print(getattr(person, "name")) # 张三
setattr(person, "age", 26)
print(person.age) # 26
print(callable(person.greet)) # True
反射机制使代码能够在运行时动态地操作对象，这对于构建通用框架、ORM系统、序列化框架等非常有用。
元类是类的类，用于创建类的工厂。通过元类，我们可以在类定义时动态修改类的行为。
# 简单的元类示例
class Meta(type):
def __new__(cls, name, bases, attrs):
# 为每个方法添加前缀
for key, value in attrs.items():
if callable(value) and not key.startswith("_"):
attrs[key] = lambda self, *args, **kwargs: value(self, *args, **kwargs)
return super().__new__(cls, name, bases, attrs)
class MyClass(metaclass=Meta):
def method1(self):
return "方法1"
def method2(self):
return "方法2"
obj = MyClass()
print(obj.method1()) # 方法1
元类在框架开发中非常有用，例如Django的ORM系统就使用元类来实现模型类的自动注册。使用元类可以拦截类的创建过程，在类创建之前修改其属性或行为。
描述符是实现__get__、__set__或__delete__方法的对象，用于管理属性的访问。
class PropertyDescriptor:
"""属性描述符示例"""
def __init__(self, name):
self.name = name
def __get__(self, obj, objtype=None):
return getattr(obj, f"_{self.name}", None)
def __set__(self, obj, value):
if value < 0:
raise ValueError("值不能为负数")
setattr(obj, f"_{self.name}", value)
def __delete__(self, obj):
raise AttributeError("不能删除此属性")
class Rectangle:
width = PropertyDescriptor("width")
height = PropertyDescriptor("height")
def __init__(self, width, height):
self.width = width
self.height = height
def area(self):
return self.width * self.height
rect = Rectangle(5, 3)
print(rect.area()) # 15
rect.width = 10
print(rect.area()) # 30
rect.width = -5 # 抛出ValueError
属性描述符提供了一种强大的方式来定义属性的行为，实现数据验证、计算属性、只读属性等高级功能。
在某些高级场景下，需要动态创建类或函数。
# 动态创建函数
def create_adder(n):
def adder(x):
return x + n
return adder
add5 = create_adder(5)
print(add5(10)) # 15
# 动态创建类
def create_class(name):
return type(name, (), {"value": 10})
MyClass = create_class("MyClass")
obj = MyClass()
print(obj.value) # 10
Python的threading模块提供了多线程编程的支持。多线程适用于I/O密集型任务，如网络请求、文件读写等。
import threading
import time
def worker(name, delay):
print(f"线程 {name} 开始")
time.sleep(delay)
print(f"线程 {name} 结束")
# 创建线程
threads = []
for i in range(3):
t = threading.Thread(target=worker, args=(f"Thread-{i}", i))
threads.append(t)
t.start()
# 等待所有线程完成
for t in threads:
t.join()
print("所有线程执行完毕")
线程的主要优势在于能够并发执行任务，提高程序的响应性。对于I/O密集型任务，多线程可以显著提高效率，因为线程在等待I/O操作时可以释放GIL（全局解释器锁），允许其他线程执行。
当多个线程共享数据时，需要使用同步机制来避免竞态条件。
| 机制 | 说明 |
|---|---|
| Lock | 互斥锁，同一时刻只允许一个线程访问 |
| RLock | 可重入锁，同一线程可多次获取 |
| Semaphore | 信号量，控制同时访问的资源数量 |
| Event | 事件，用于线程间通信 |
| Condition | 条件变量，用于复杂的线程同步 |
| Queue | 队列，生产者-消费者模式 |
import threading
# 使用Lock保护共享资源
counter = 0
lock = threading.Lock()
def increment():
global counter
for _ in range(100000):
with lock:
counter += 1
threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"最终计数器值: {counter}") # 500000（正确）
在这个例子中，如果不用Lock保护，counter += 1操作可能会被多个线程同时执行，导致计数不准确。Lock确保了同一时刻只有一个线程能够修改counter。
由于Python的GIL限制，CPU密集型任务应该使用多进程来实现真正的并行计算。multiprocessing模块提供了多进程编程的支持。
import multiprocessing
import time
def cpu_task(n):
"""CPU密集型任务"""
result = sum(i*i for i in range(n))
return result
if __name__ == "__main__":
# 创建进程池
with multiprocessing.Pool(processes=4) as pool:
# 并行执行任务
results = pool.map(cpu_task, [1000000] * 8)
print(f"结果: {results}")
多进程每个进程有独立的GIL，可以真正并行执行。multiprocessing模块的接口与threading模块类似，但使用的是进程而不是线程。进程间的通信需要使用Queue、Pipe等机制。
import multiprocessing
def worker(queue):
"""从队列获取数据并处理"""
while not queue.empty():
item = queue.get()
print(f"处理: {item}")
queue.task_done()
if __name__ == "__main__":
# 创建队列
queue = multiprocessing.Queue()
# 添加任务
for i in range(5):
queue.put(i)
# 启动工作进程
processes = []
for _ in range(2):
p = multiprocessing.Process(target=worker, args=(queue,))
p.start()
processes.append(p)
# 等待完成
for p in processes:
p.join()
print("所有任务处理完毕")
异步编程是一种并发编程范式，允许在等待I/O操作时执行其他任务。Python 3.5+引入了async/await语法，使得异步编程更加简洁。
asyncio是Python的异步编程库，提供了一个事件循环来管理异步任务的执行。
import asyncio
async def fetch_data(url):
"""模拟异步获取数据"""
print(f"开始获取: {url}")
await asyncio.sleep(2) # 模拟网络请求
print(f"完成获取: {url}")
return f"数据-{url}"
async def main():
# 异步任务列表
urls = ["url1", "url2", "url3"]
# 并发执行多个异步任务
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"所有结果: {results}")
# 运行异步程序
asyncio.run(main())
异步编程的核心优势在于能够高效地处理大量并发I/O操作，例如网络请求、数据库查询等。与多线程相比，异步编程不需要切换线程的开销，代码更加简洁易读。
async def用于定义异步函数，await用于等待异步操作完成。
import asyncio
async def async_function():
"""异步函数"""
print("异步函数开始")
result = await another_async_function()
print("异步函数结束")
return result
async def another_async_function():
"""另一个异步函数"""
await asyncio.sleep(1)
return "异步结果"
# 运行
result = asyncio.run(async_function())
await只能在async函数中使用，它暂停当前协程的执行，等待另一个协程完成后再继续执行。这使得我们能够以同步的方式编写异步代码，大大提高了代码的可读性。
import asyncio
class AsyncIterator:
"""异步迭代器"""
def __init__(self, data):
self.data = data
self.index = 0
def __aiter__(self):
return self
async def __anext__(self):
if self.index >= len(self.data):
raise StopAsyncIteration
value = self.data[self.index]
self.index += 1
await asyncio.sleep(0.1) # 模拟异步操作
return value
async def main():
async for item in AsyncIterator([1, 2, 3, 4, 5]):
print(item)
asyncio.run(main())
异步迭代器允许在迭代过程中执行异步操作，这对于处理流式数据特别有用，例如异步读取文件、异步网络请求等场景。
在异步编程中，aiohttp是常用的异步HTTP客户端库。
import asyncio
import aiohttp
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
urls = [
"https://httpbin.org/get",
"https://httpbin.org/ip",
"https://httpbin.org/headers"
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for url, result in zip(urls, results):
print(f"{url}: {len(result)} bytes")
asyncio.run(main())
aiohttp能够并发地发起多个HTTP请求，充分利用异步I/O的优势，比同步HTTP客户端效率高得多。
SQLite是Python内置的轻量级数据库，不需要额外的服务器进程，适合小型项目和快速原型开发。
import sqlite3
# 连接到数据库（不存在则创建）
conn = sqlite3.connect("example.db")
cursor = conn.cursor()
# 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
age INTEGER
)
""")
# 插入数据
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
("张三", "zhangsan@example.com", 25))
# 批量插入
users = [
("李四", "lisi@example.com", 30),
("王五", "wangwu@example.com", 28)
]
cursor.executemany("INSERT INTO users (name, email, age) VALUES (?, ?, ?)", users)
# 查询数据
cursor.execute("SELECT * FROM users WHERE age > ?", (25,))
results = cursor.fetchall()
for row in results:
print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}, Age: {row[3]}")
# 更新数据
cursor.execute("UPDATE users SET age = ? WHERE name = ?", (26, "张三"))
# 删除数据
cursor.execute("DELETE FROM users WHERE name = ?", ("王五",))
# 提交更改并关闭连接
conn.commit()
conn.close()
Python的DB-API 2.0是数据库操作的标准化接口，主要方法包括connect()、cursor()、execute()、fetchone()、fetchall()、close()等。
| 方法 | 说明 |
|---|---|
| connect() | 建立数据库连接 |
| cursor() | 创建游标对象 |
| execute() | 执行SQL语句 |
| executemany() | 批量执行SQL |
| fetchone() | 获取一条记录 |
| fetchall() | 获取所有记录 |
| commit() | 提交事务 |
| rollback() | 回滚事务 |
| close() | 关闭连接 |
ORM（Object-Relational Mapping）将对象映射到数据库表，简化数据库操作。SQLAlchemy是Python最流行的ORM框架之一。
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
# 创建引擎
engine = create_engine("sqlite:///example.db", echo=True)
# 定义模型基类
Base = declarative_base()
# 定义模型类
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
email = Column(String(50), unique=True)
age = Column(Integer)
# 创建表
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 插入
new_user = User(name="张三", email="zhangsan@example.com", age=25)
session.add(new_user)
session.commit()
# 查询
user = session.query(User).filter_by(name="张三").first()
print(f"查询结果: {user.name}, {user.email}")
# 更新
user.age = 26
session.commit()
# 删除
session.delete(user)
session.commit()
session.close()
socket是网络通信的基础，Python的socket模块提供了底层的网络编程接口。
import socket
# 服务器端
def echo_server(host="127.0.0.1", port=8888):
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((host, port))
server_socket.listen(5)
print(f"服务器监听 {host}:{port}")
while True:
client_socket, addr = server_socket.accept()
print(f"收到来自 {addr} 的连接")
# 接收数据
data = client_socket.recv(1024)
print(f"收到数据: {data.decode()}")
# 发送响应
client_socket.sendall(b"服务器收到了你的消息")
client_socket.close()
# 客户端
def echo_client(host="127.0.0.1", port=8888):
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect((host, port))
message = "Hello, Server!"
client_socket.sendall(message.encode())
response = client_socket.recv(1024)
print(f"服务器响应: {response.decode()}")
client_socket.close()
socket编程需要处理粘包、半关闭、并发等问题，在生产环境中通常使用更高级的框架。socket.AF_INET表示IPv4，socket.SOCK_STREAM表示TCP协议。
Python的http.server模块提供了简单的HTTP服务器实现。
from http.server import HTTPServer, BaseHTTPRequestHandler
class SimpleHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/html; charset=utf-8")
self.end_headers()
response = """
<!DOCTYPE html>
<html>
<head><title>简单HTTP服务器</title></head>
<body>
<h1>欢迎访问</h1>
<p>这是一个简单的HTTP服务器</p>
</body>
</html>
"""
self.wfile.write(response.encode("utf-8"))
def log_message(self, format, *args):
print(f"{self.address_string()} - {format % args}")
# 启动服务器
server_address = ("", 8000)
httpd = HTTPServer(server_address, SimpleHandler)
print("服务器运行在 http://localhost:8000")
httpd.serve_forever()
unittest是Python内置的单元测试框架。
import unittest
def add(a, b):
return a + b
def divide(a, b):
if b == 0:
raise ValueError("除数不能为零")
return a / b
class TestMathFunctions(unittest.TestCase):
def test_add(self):
self.assertEqual(add(1, 2), 3)
self.assertEqual(add(-1, 1), 0)
self.assertEqual(add(0, 0), 0)
def test_divide(self):
self.assertEqual(divide(10, 2), 5)
self.assertEqual(divide(9, 3), 3)
def test_divide_by_zero(self):
with self.assertRaises(ValueError):
divide(10, 0)
if __name__ == "__main__":
unittest.main()
unittest提供了丰富的断言方法，包括assertEqual、assertTrue、assertFalse、assertIsNone、assertIsNotNone、assertIn、assertRaises等，可以满足大多数测试需求。
pytest是更现代、更简洁的测试框架，兼容unittest。
# test_example.py
import pytest
def add(a, b):
return a + b
# 测试函数以test_开头
def test_add():
assert add(1, 2) == 3
assert add(-1, 1) == 0
assert add(0, 0) == 0
# 使用fixture
@pytest.fixture
def sample_data():
return [1, 2, 3, 4, 5]
def test_sample_data(sample_data):
assert len(sample_data) == 5
assert sum(sample_data) == 15
# 参数化测试
@pytest.mark.parametrize("a,b,expected", [
(1, 2, 3),
(0, 0, 0),
(-1, 1, 0),
])
def test_add_parametrized(a, b, expected):
assert add(a, b) == expected
pytest的fixture机制提供了强大的测试数据管理和依赖注入功能，parametrize装饰器可以轻松实现参数化测试，大大减少测试代码的重复。
import pdb
def buggy_function(x):
result = x * 2
# 设置断点
pdb.set_trace()
return result / 0 # 这里会出错
# 使用pdb命令调试：
# n (next) - 执行下一行
# s (step) - 进入函数
# c (continue) - 继续执行
# p variable - 打印变量值
# l (list) - 查看当前代码
# q (quit) - 退出调试
Python还提供了其他调试工具，如ipdb（交互式pdb）、traceback模块（查看异常追溯）、logging模块（日志记录）等。
PEP 8是Python的官方代码风格指南，主要规则包括：
| 规则 | 说明 |
|---|---|
| 缩进 | 使用4个空格，不要使用Tab |
| 行长度 | 最多79个字符 |
| 空行 | 类之间两个空行，方法之间一个空行 |
| 导入 | 按标准库、第三方库、本地模块顺序分组 |
| 命名 | 函数和变量用snake_case，类名用PascalCase |
| 空格 | 避免不必要的空格 |
可以使用工具自动检查代码规范：pylint、flake8、black（自动格式化）。
设计模式是软件工程中常见问题的通用解决方案。
class Singleton:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
s1 = Singleton()
s2 = Singleton()
print(s1 is s2) # True
class Dog:
def speak(self):
return "Woof!"
class Cat:
def speak(self):
return "Meow!"
class AnimalFactory:
@staticmethod
def create_animal(animal_type):
animals = {"dog": Dog, "cat": Cat}
return animals.get(animal_type.lower())()
animal = AnimalFactory.create_animal("dog")
print(animal.speak()) # Woof!
class Subject:
def __init__(self):
self._observers = []
def attach(self, observer):
self._observers.append(observer)
def detach(self, observer):
self._observers.remove(observer)
def notify(self):
for observer in self._observers:
observer.update()
class Observer:
def update(self):
print("收到通知")
# 创建虚拟环境
python -m venv venv
# 激活虚拟环境
# Windows
venv\Scripts\activate
# Linux/Mac
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt
# 导出依赖
pip freeze > requirements.txt
本教程涵盖了Python进阶的核心知识，包括：
掌握这些进阶知识后，您可以更好地构建复杂的Python应用程序，处理高性能、高并发的场景。持续学习和实践是提升Python技能的关键。