小学生必须掌握的Python语法与代码示例

目录

• 📘 第一阶段：基础核心
• 📙 第二阶段：进阶核心
• 📗 第三阶段：高级精通
• 💡 如何有效提升？
• 一、Python基础语法
• 1.1 变量与赋值
• 1.2 注释与代码结构
• 二、Python核心数据类型
• 2.1 数字类型
• 2.2 序列类型对比与操作
• 2.3 字典与集合
• 三、流程控制
• 3.1 条件语句
• 3.2 循环结构
• 四、函数详解
• 4.1 函数定义与调用
• 4.2 参数类型详解
• 4.3 高阶函数与lambda
• 4.4 作用域与闭包
• 五、综合实例：学生成绩管理系统
• 学习建议
• 一、面向对象编程高级特性
• 1.1 类方法与静态方法
• 1.2 属性装饰器与描述符
• 二、装饰器高级应用
• 2.1 参数化装饰器
• 2.2 类装饰器
• 三、上下文管理器与资源管理
• 3.1 自定义上下文管理器
• 四、元类与元编程
• 4.1 自定义元类
• 五、并发与并行编程
• 5.1 多线程与线程安全
• 5.2 异步编程(asyncio)
• 六、性能优化与元编程
• 6.1 使用__slots__减少内存
• 学习建议

学习Python从入门到精通，关键在于先建立扎实的核心概念（如基础语法、函数、面向对象），再系统掌握其高级特性和生态工具。下面我为你规划了一条清晰的学习路径，并提供了核心知识框架和示例。

这个阶段的目标是学会用Python表达基本逻辑和操作数据。

• 基础语法与环境
  • 核心：理解变量、基本数据类型（整型、浮点型、布尔型）、运算符和注释。
  • 示例：快速体验Python的交互模式。

# 变量与运算
price = 19.95
quantity = 3
total = price * quantity
print(f"总价: {total}")  # 输出: 总价: 59.85
# 类型转换与检查
num_str = "123"
num_int = int(num_str)
print(isinstance(num_int, int))  # 输出: True

• 核心数据结构
  • Python的强大很大程度上源于其灵活的内置数据结构。下表对比了它们的关键特性：

类型	可变性	是否有序	元素要求	典型用途与示例
列表(List)	可变	有序	可重复，任意类型	存储有序序列，内容可增删改。tasks = ['写报告', '开会', 1]
元组(Tuple)	不可变	有序	可重复，任意类型	存储不可变序列，常用于保证数据安全或作为字典键。point = (10, 20)
字典(Dict)	可变	无序(按key存取)	Key不可重复且不可变	存储键值对，实现快速查询。student = {'name': '小明', 'score': 90}
集合(Set)	可变	无序	不可重复	去重、集合运算（交集、并集）。unique_numbers = {1, 2, 2, 3} # 结果{1, 2, 3}

示例：列表和字典的常用操作。

# 列表操作
fruits = ['apple', 'banana']
fruits.append('orange')  # 增加
fruits[1] = 'grape'      # 修改
last_fruit = fruits.pop() # 删除并返回最后一个元素
print(fruits)  # 输出: ['apple', 'grape']
# 字典操作
scores = {'Math': 85, 'English': 92}
scores['Science'] = 88   # 增加键值对
print(scores['Math'])    # 通过key访问: 85
for subject, score in scores.items(): # 遍历
    print(f"{subject}: {score}")

• 程序流程控制
  • 核心：使用 if-elif-else 进行条件分支，用 for 和 while 进行循环，并用 try-except 处理异常。
  • 示例：结合数据结构的综合流程控制。

# 条件与循环处理考试成绩
grade_dict = {'张三': 78, '李四': 92, '王五': 58}
for name, score in grade_dict.items():
    if score >= 90:
        level = '优秀'
    elif score >= 60:
        level = '及格'
    else:
        level = '不及格'
    print(f"{name}: {score}分 -> {level}")
# 异常处理
try:
    user_input = int(input("请输入一个数字: "))
    result = 100 / user_input
except ValueError:
    print("输入的不是有效数字！")
except ZeroDivisionError:
    print("除数不能为零！")
else:
    print(f"结果是: {result}")

掌握如何组织代码、抽象问题，并处理外部数据。

• 函数与代码复用
  • 核心：使用 def 定义函数，理解参数传递（位置参数、默认参数、可变参数 *args 和关键字参数 **kwargs），以及变量作用域。
  • 关键点：Python中，不可变对象（如整数、字符串、元组）是“按值传递”（实际是传递对象的引用，但无法修改原对象），可变对象（如列表、字典）是“按引用传递”（在函数内修改会影响原对象）。
• 示例：

# 定义带默认参数和类型提示的函数
def greet(name: str, greeting: str = "Hello") -> str:
    """返回一个问候字符串。"""
    return f"{greeting}, {name}!"
print(greet("Alice"))  # 使用默认参数
print(greet("Bob", "Hi"))  # 提供参数
# 可变参数示例：计算任意个数的和
def dynamic_sum(*args, **kwargs):
    normal_sum = sum(args)
    print(f"位置参数和: {normal_sum}")
    print(f"关键字参数: {kwargs}")
dynamic_sum(1, 2, 3, name='Tom', age=10)

• 面向对象编程 (OOP)
  • 核心：理解类(Class)与对象(Object)。掌握OOP三大特性：封装（隐藏内部细节）、继承（实现代码复用）、多态（同一接口不同实现）。
  • 示例：

class Animal:
    def __init__(self, name):  # 构造方法
        self.name = name  # 实例变量
    def speak(self):  # 方法
        raise NotImplementedError("子类必须实现此方法")
class Dog(Animal):  # 继承
    def speak(self):  # 多态：重写父类方法
        return f"{self.name} says: Woof!"
my_dog = Dog("Buddy")
print(my_dog.speak())  # 输出: Buddy says: Woof!

• 文件与数据持久化
  • 核心：使用 open() 函数读写文本和二进制文件。用 with 语句管理资源可自动关闭文件。处理 JSON、CSV 等格式数据是日常必备技能。
  • 示例：

# 使用with安全地读写文件
with open('note.txt', 'w', encoding='utf-8') as f:
    f.write("Hello, World!n这是第二行。")
with open('note.txt', 'r', encoding='utf-8') as f:
    content = f.read()
    print(content)
# 处理JSON数据（常用于API和配置）
import json
data = {'name': '小明', 'hobbies': ['阅读', '游泳']}
json_str = json.dumps(data, ensure_ascii=False)  # 转为JSON字符串
loaded_data = json.loads(json_str)  # 解析JSON字符串

深入语言特性和生态，解决复杂工程问题。

• 常用高级特性
  • 装饰器：在不修改原函数代码的情况下，为函数添加额外功能（如日志、计时、权限检查）。

def timer(func):
    import time
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 执行耗时: {end-start:.2f}秒")
        return result
    return wrapper
@timer
def some_task():
    time.sleep(1)
some_task()

• 生成器：使用 yield 关键字，用于惰性生成大量数据，节省内存。

def fibonacci(limit):
    a, b = 0, 1
    while a < limit:
        yield a  # 每次产生一个值，函数状态会暂停保留
        a, b = b, a + b
for num in fibonacci(10):
    print(num)  # 输出: 0 1 1 2 3 5 8

• 上下文管理器：除了文件操作，可以自定义 __enter__ 和 __exit__ 方法来管理资源（如网络连接、数据库连接）。

1. 主流应用领域与库

   • Web开发：使用 Flask（轻量灵活）或 Django（功能全面）框架。
   • 数据分析与科学计算：NumPy（数组计算）、Pandas（数据分析）、Matplotlib（数据可视化）。
   • 人工智能与机器学习：Scikit-learn（传统机器学习）、TensorFlow/PyTorch（深度学习）。
   • 自动化与脚本：使用 os、sys、subprocess 等标准库进行文件管理、系统操作。

2. 性能优化与工程化

   • 性能分析：使用 cProfile、timeit 模块分析代码瓶颈。
   • 调试与测试：使用 pdb 进行调试，用 unittest 或 pytest 框架编写单元测试。
   • 代码规范：遵循 PEP 8 风格指南，使用 Black、isort 等工具自动化格式化。

• 遵循路径：建议按上述三个阶段顺序学习，每个阶段打好基础再进入下一个。
• 实践至上：学习每个知识点后，立刻动手编写代码。可以从书中的例子、小型脚本（如自动化整理文件、爬取天气数据）开始。
• 阅读优秀代码：在GitHub上阅读热门项目的源码，学习代码组织方式和最佳实践。
• 参与项目：找一个小型但完整的项目进行实践，例如用Flask搭建一个博客，或用Pandas分析一份数据集。这是将知识融会贯通的最佳方式。

如果你想深入了解某个特定领域（比如Web框架的详细对比，或者数据科学库的具体入门方法），可以在网上搜索更具体的资料和学习建议。

我将通过递进的示例，系统讲解Python的核心语法、数据类型和函数，帮助你从基础到进阶。

# 变量命名规范
name = "Python"          # 字符串
version = 3.9           # 整数
rating = 9.5           # 浮点数
is_awesome = True      # 布尔值
# 多重赋值
x, y, z = 1, 2, 3
# 链式赋值
a = b = c = 100
# 变量交换（Python特有）
m, n = 10, 20
m, n = n, m  # 交换后 m=20, n=10

# 单行注释
"""
多行注释（实际是三引号字符串）
通常用于模块、函数说明
"""
# 代码块通过缩进定义
if True:
    print("缩进4个空格")  # 属于if代码块
    print("同一代码块")
print("已退出代码块")     # 不属于if代码块

# 整数
int_num = 42          # 十进制
binary_num = 0b1010   # 二进制: 10
hex_num = 0xFF        # 十六进制: 255
# 浮点数
float_num = 3.14159
scientific = 1.23e-4  # 0.000123
# 复数
complex_num = 3 + 4j
# 数值运算
print(10 / 3)     # 浮点除法: 3.333...
print(10 // 3)    # 整除: 3
print(10 % 3)     # 取模: 1
print(2 ** 3)     # 幂运算: 8

下表展示了Python主要序列类型的关键特性：

类型	可变性	创建方式	特点	使用场景
列表	可变	[1, 2, 3] 或 list()	有序，元素可重复，可包含不同类型	存储可变数据集，如待办事项
元组	不可变	(1, 2, 3) 或 tuple()	有序，创建后不能修改	固定数据集合，如坐标、配置项
字符串	不可变	"hello" 或 str()	字符序列，支持多种操作	文本处理，格式化输出
范围	不可变	range(5)	生成数字序列，内存高效	循环计数，序列生成

# 列表操作
fruits = ['apple', 'banana', 'orange']
fruits.append('grape')      # 末尾添加
fruits.insert(1, 'pear')    # 指定位置插入
fruits.remove('banana')     # 删除元素
popped = fruits.pop()       # 移除并返回最后一个
print(fruits[0:2])          # 切片: ['apple', 'pear']
# 列表推导式（高效创建列表）
squares = [x**2 for x in range(10) if x % 2 == 0]
# 结果: [0, 4, 16, 36, 64]
# 元组
coordinates = (10.5, 20.3)
x, y = coordinates  # 解包
single_tuple = (42,)  # 单元素元组需要逗号
# 字符串操作
text = "Python Programming"
print(text.upper())         # 转大写
print(text.find("Pro"))     # 查找位置: 7
print(text.split(" "))      # 分割: ['Python', 'Programming']
print("Hello, {}!".format(name))  # 格式化
print(f"Version: {version}")      # f-string (Python 3.6+)

# 字典（键值对集合）
student = {
    "name": "Alice",
    "age": 20,
    "courses": ["Math", "Physics"]
}
student["grade"] = "A"          # 添加键值对
print(student.get("age"))       # 安全获取: 20
print(student.get("score", "N/A"))  # 默认值: N/A
# 遍历字典
for key, value in student.items():
    print(f"{key}: {value}")
# 字典推导式
squared_dict = {x: x**2 for x in range(5)}
# 结果: {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}
# 集合（无序不重复元素）
set_a = {1, 2, 3, 3, 4}  # 自动去重: {1, 2, 3, 4}
set_b = {3, 4, 5, 6}
print(set_a | set_b)   # 并集: {1, 2, 3, 4, 5, 6}
print(set_a & set_b)   # 交集: {3, 4}
print(set_a - set_b)   # 差集: {1, 2}

# if-elif-else结构
score = 85
if score >= 90:
    grade = "A"
elif score >= 80:
    grade = "B"
elif score >= 70:
    grade = "C"
else:
    grade = "D"
# 三元表达式
status = "及格" if score >= 60 else "不及格"
# 多条件判断
age = 25
is_student = True
if 18 <= age <= 30 and is_student:
    discount = 0.5
elif age > 60 or not is_student:
    discount = 0.2
else:
    discount = 0

# for循环遍历序列
colors = ["red", "green", "blue"]
for i, color in enumerate(colors):  # enumerate获取索引和值
    print(f"颜色{i}: {color}")
# 遍历字典
for key in student.keys():
    print(key)
# while循环
count = 0
while count < 5:
    print(f"计数: {count}")
    count += 1
    if count == 3:
        break  # 跳出循环
# 循环控制
for i in range(10):
    if i % 2 == 0:
        continue  # 跳过偶数
    print(f"奇数: {i}")
# 循环中的else子句（循环正常结束时执行）
for i in range(3):
    print(i)
else:
    print("循环完成")

def greet(name, greeting="Hello"):
    """
    问候函数
    参数:
    name: 姓名
    greeting: 问候语，默认为'Hello'
    返回:
    问候字符串
    """
    return f"{greeting}, {name}!"
# 函数调用
print(greet("Alice"))                # 位置参数
print(greet(greeting="Hi", name="Bob"))  # 关键字参数
# 返回多个值（实际返回元组）
def min_max(numbers):
    return min(numbers), max(numbers)
low, high = min_max([3, 1, 4, 1, 5])
print(f"最小值: {low}, 最大值: {high}")

def flexible_func(a, b=10, *args, **kwargs):
    """
    演示各种参数类型
    a: 位置参数（必须）
    b: 默认参数（可选）
    *args: 可变位置参数（元组）
    **kwargs: 可变关键字参数（字典）
    """
    print(f"a: {a}, b: {b}")
    print(f"args: {args}")
    print(f"kwargs: {kwargs}")
# 调用示例
flexible_func(1)                     # a:1, b:10, args:(), kwargs:{}
flexible_func(1, 2, 3, 4, x=5, y=6)  # a:1, b:2, args:(3,4), kwargs:{'x':5,'y':6}
# 参数解包
params = {'name': 'Charlie', 'age': 25}
def person_info(name, age):
    return f"{name} is {age} years old"
print(person_info(**params))  # 字典解包

# lambda表达式（匿名函数）
add = lambda x, y: x + y
print(add(5, 3))  # 8
# 常用于排序
students = [
    {'name': 'Alice', 'score': 85},
    {'name': 'Bob', 'score': 92},
    {'name': 'Charlie', 'score': 78}
]
# 按分数排序
sorted_students = sorted(students, key=lambda s: s['score'], reverse=True)
# map函数：对序列每个元素应用函数
numbers = [1, 2, 3, 4]
squared = list(map(lambda x: x**2, numbers))  # [1, 4, 9, 16]
# filter函数：过滤序列
evens = list(filter(lambda x: x % 2 == 0, numbers))  # [2, 4]
# 函数作为返回值
def make_multiplier(n):
    def multiplier(x):
        return x * n
    return multiplier
double = make_multiplier(2)
print(double(5))  # 10

# 全局变量与局部变量
global_var = "全局"
def scope_test():
    local_var = "局部"
    print(global_var)  # 可访问全局变量
    print(local_var)   # 可访问局部变量
    # 修改全局变量需要声明
    global global_var
    global_var = "已修改"
scope_test()
# print(local_var)  # 错误！不能访问函数内的局部变量
# 闭包：函数记住其外部作用域
def counter():
    count = 0
    def increment():
        nonlocal count  # 声明非局部变量
        count += 1
        return count
    return increment
my_counter = counter()
print(my_counter())  # 1
print(my_counter())  # 2 - count被记住了

class StudentManager:
    """学生成绩管理类"""
    def __init__(self):
        self.students = {}
    def add_student(self, name, scores):
        """添加学生及其成绩"""
        self.students[name] = scores
    def calculate_average(self, name):
        """计算学生平均分"""
        if name not in self.students:
            return None
        scores = self.students[name]
        return sum(scores) / len(scores) if scores else 0
    def get_top_student(self):
        """获取平均分最高的学生"""
        if not self.students:
            return None
        # 使用lambda和max函数
        return max(self.students.items(), 
                  key=lambda item: self.calculate_average(item[0]))
    def analyze_scores(self):
        """成绩分析"""
        all_scores = []
        for scores in self.students.values():
            all_scores.extend(scores)
        if not all_scores:
            return "暂无数据"
        analysis = {
            'total_students': len(self.students),
            'total_scores': len(all_scores),
            'average_all': sum(all_scores) / len(all_scores),
            'max_score': max(all_scores),
            'min_score': min(all_scores)
        }
        return analysis
# 使用示例
manager = StudentManager()
# 添加学生数据
manager.add_student("Alice", [85, 92, 78])
manager.add_student("Bob", [76, 88, 91])
manager.add_student("Charlie", [92, 95, 89])
# 分析数据
print(f"Alice的平均分: {manager.calculate_average('Alice'):.2f}")
top_student = manager.get_top_student()
print(f"最高分学生: {top_student[0]}, 平均分: {manager.calculate_average(top_student[0]):.2f}")
analysis = manager.analyze_scores()
print(f"总学生数: {analysis['total_students']}")
print(f"所有科目平均分: {analysis['average_all']:.2f}")

1. 练习顺序：先掌握基础语法和数据类型，再深入函数和高级特性
2. 实践方法：每个代码示例都手动输入并修改，观察不同变化
3. 项目驱动：尝试用所学知识解决实际问题，如数据分析小脚本、自动化工具
4. 调试技巧：使用print()调试，逐步构建复杂功能

这个知识体系覆盖了Python核心概念的80%，掌握后即可编写实用的Python程序。建议按照示例顺序实践，逐步建立编程思维。

class Date:
    """日期类演示类方法与静态方法"""
    def __init__(self, year, month, day):
        self.year = year
        self.month = month
        self.day = day
    # 实例方法 - 操作实例属性
    def display(self):
        return f"{self.year}-{self.month:02d}-{self.day:02d}"
    @classmethod  # 类方法 - 操作类本身
    def from_string(cls, date_string):
        """从字符串创建Date实例"""
        year, month, day = map(int, date_string.split('-'))
        return cls(year, month, day)  # cls代表类本身
    @classmethod
    def is_leap_year(cls, year):
        """判断是否为闰年"""
        return year % 400 == 0 or (year % 4 == 0 and year % 100 != 0)
    @staticmethod  # 静态方法 - 工具函数，不依赖类或实例
    def days_in_month(month, year):
        """返回指定月份的天数"""
        if month in [1, 3, 5, 7, 8, 10, 12]:
            return 31
        elif month in [4, 6, 9, 11]:
            return 30
        elif month == 2:
            return 29 if Date.is_leap_year(year) else 28
        else:
            raise ValueError("无效月份")
# 使用示例
d1 = Date(2024, 5, 20)
print(d1.display())  # 实例方法
d2 = Date.from_string("2024-05-21")  # 类方法作为替代构造器
print(d2.display())
print(Date.is_leap_year(2024))  # True
print(Date.days_in_month(2, 2024))  # 29

class Temperature:
    """使用属性装饰器控制属性访问"""
    def __init__(self, celsius=0):
        self._celsius = celsius  # 私有属性
    @property
    def celsius(self):
        """获取摄氏温度"""
        print("获取摄氏温度")
        return self._celsius
    @celsius.setter
    def celsius(self, value):
        """设置摄氏温度"""
        if value < -273.15:
            raise ValueError("温度不能低于绝对零度(-273.15°C)")
        print(f"设置摄氏温度为: {value}")
        self._celsius = value
    @property
    def fahrenheit(self):
        """计算华氏温度（只读属性）"""
        return self._celsius * 9/5 + 32
    @fahrenheit.setter
    def fahrenheit(self, value):
        """通过华氏温度设置摄氏温度"""
        self._celsius = (value - 32) * 5/9
# 描述符类
class ValidatedAttribute:
    """自定义描述符验证属性值"""
    def __init__(self, min_value=None, max_value=None):
        self.min_value = min_value
        self.max_value = max_value
        self.data = {}
    def __get__(self, instance, owner):
        if instance is None:
            return self
        return self.data.get(id(instance))
    def __set__(self, instance, value):
        if self.min_value is not None and value < self.min_value:
            raise ValueError(f"值不能小于 {self.min_value}")
        if self.max_value is not None and value > self.max_value:
            raise ValueError(f"值不能大于 {self.max_value}")
        self.data[id(instance)] = value
class Product:
    price = ValidatedAttribute(min_value=0, max_value=10000)
    quantity = ValidatedAttribute(min_value=0, max_value=1000)
    def __init__(self, name, price, quantity):
        self.name = name
        self.price = price  # 触发描述符的__set__
        self.quantity = quantity
# 使用示例
temp = Temperature(25)
print(f"摄氏: {temp.celsius}°C")  # 触发@property
print(f"华氏: {temp.fahrenheit}°F")
temp.celsius = 30  # 触发@celsius.setter
temp.fahrenheit = 100  # 触发@fahrenheit.setter
try:
    p = Product("手机", -100, 10)
except ValueError as e:
    print(f"错误: {e}")  # 值不能小于 0
p2 = Product("笔记本", 5000, 5)
print(f"{p2.name}: 价格{p2.price}, 库存{p2.quantity}")

import time
from functools import wraps
from typing import Callable, Any
def retry(max_attempts: int = 3, delay: float = 1.0):
    """
    参数化重试装饰器
    Args:
        max_attempts: 最大重试次数
        delay: 重试间隔(秒)
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    print(f"尝试第 {attempt + 1} 次...")
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        print(f"失败，{delay}秒后重试...")
                        time.sleep(delay)
            print(f"所有 {max_attempts} 次尝试均失败")
            raise last_exception
        return wrapper
    return decorator
# 创建缓存装饰器
def cache(maxsize: int = 128):
    """带LRU缓存的参数化装饰器"""
    def decorator(func: Callable) -> Callable:
        cache_dict = {}
        cache_keys = []
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 创建缓存键
            key = (args, frozenset(kwargs.items()))
            if key in cache_dict:
                print(f"缓存命中: {func.__name__}{args}")
                return cache_dict[key]
            print(f"计算: {func.__name__}{args}")
            result = func(*args, **kwargs)
            # LRU缓存逻辑
            cache_dict[key] = result
            cache_keys.append(key)
            if len(cache_dict) > maxsize:
                oldest_key = cache_keys.pop(0)
                del cache_dict[oldest_key]
            return result
        wrapper.clear_cache = lambda: (cache_dict.clear(), cache_keys.clear())
        return wrapper
    return decorator
# 使用参数化装饰器
@retry(max_attempts=5, delay=0.5)
def unstable_api_call():
    """模拟不稳定的API调用"""
    import random
    if random.random() < 0.7:
        raise ConnectionError("API连接失败")
    return "API调用成功"
@cache(maxsize=2)
def expensive_computation(n):
    """模拟昂贵计算"""
    print(f"执行昂贵计算: {n}")
    time.sleep(1)
    return n * n
# 测试
try:
    result = unstable_api_call()
    print(result)
except Exception as e:
    print(f"最终失败: {e}")
print(expensive_computation(5))  # 计算
print(expensive_computation(5))  # 从缓存获取
print(expensive_computation(3))  # 计算
print(expensive_computation(4))  # 计算，触发LRU淘汰
print(expensive_computation(5))  # 重新计算（已被淘汰）

from dataclasses import dataclass, field
from typing import List, ClassVar
import json
# 类装饰器：为类添加单例模式
def singleton(cls):
    """单例模式装饰器"""
    instances = {}
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance
@singleton
class AppConfig:
    def __init__(self):
        self.settings = {"theme": "dark", "language": "zh-CN"}
    def update(self, key, value):
        self.settings[key] = value
# 使用@dataclass自动生成方法
@dataclass(order=True, frozen=False)  # frozen=True创建不可变实例
class Person:
    """使用dataclass自动生成__init__, __repr__, __eq__等方法"""
    name: str
    age: int
    email: str = ""  # 默认值
    hobbies: List[str] = field(default_factory=list)  # 可变默认值的正确写法
    id: ClassVar[int] = 0  # 类变量
    def __post_init__(self):
        """初始化后自动调用"""
        Person.id += 1
    def to_dict(self):
        return {"name": self.name, "age": self.age, "hobbies": self.hobbies}
# 测试
config1 = AppConfig()
config2 = AppConfig()
print(f"是否是同一个实例: {config1 is config2}")  # True
config1.update("theme", "light")
print(f"config2的主题: {config2.settings['theme']}")  # light
# dataclass使用
p1 = Person("Alice", 25, "alice@example.com", ["reading", "swimming"])
p2 = Person("Bob", 30)
print(p1)  # Person(name='Alice', age=25, email='alice@example.com', hobbies=['reading', 'swimming'])
print(p1 == p2)  # False
print(p1.to_dict())  # {'name': 'Alice', 'age': 25, 'hobbies': ['reading', 'swimming']}

import sqlite3
from contextlib import contextmanager
from typing import Optional, Iterator
# 方法1：使用类实现上下文管理器
class DatabaseConnection:
    """数据库连接上下文管理器"""
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None
    def __enter__(self) -> sqlite3.Connection:
        """进入上下文时调用"""
        print(f"连接数据库: {self.db_path}")
        self.connection = sqlite3.connect(self.db_path)
        self.connection.row_factory = sqlite3.Row  # 返回字典样式的行
        return self.connection
    def __exit__(self, exc_type, exc_val, exc_tb):
        """退出上下文时调用"""
        if self.connection:
            if exc_type is None:
                print("提交事务并关闭连接")
                self.connection.commit()
            else:
                print(f"发生错误: {exc_val}，回滚事务")
                self.connection.rollback()
            self.connection.close()
        print("数据库连接已关闭")
        return False  # 不抑制异常，True表示抑制
# 方法2：使用@contextmanager装饰器
@contextmanager
def temp_table(connection: sqlite3.Connection, table_name: str) -> Iterator:
    """创建临时表的上下文管理器"""
    cursor = connection.cursor()
    try:
        print(f"创建临时表: {table_name}")
        cursor.execute(f"CREATE TEMP TABLE {table_name} (id INTEGER, name TEXT)")
        yield cursor  # 这里将控制权交给with块内的代码
    finally:
        print(f"清理临时表: {table_name}")
        cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
        cursor.close()
# 方法3：处理多个资源的上下文管理器
from contextlib import ExitStack
class MultiResourceManager:
    """管理多个资源的上下文管理器"""
    def __init__(self):
        self.stack = ExitStack()
    def add_file(self, filepath, mode='r'):
        """添加文件到资源栈"""
        file_obj = self.stack.enter_context(open(filepath, mode, encoding='utf-8'))
        return file_obj
    def add_connection(self, db_path):
        """添加数据库连接到资源栈"""
        conn = self.stack.enter_context(DatabaseConnection(db_path))
        return conn
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stack.close()
# 使用示例
print("=== 示例1: 数据库连接上下文 ===")
with DatabaseConnection(":memory:") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    with temp_table(conn, "temp_data") as temp_cursor:
        temp_cursor.execute("INSERT INTO temp_data VALUES (1, '测试')")
        temp_cursor.execute("SELECT * FROM temp_data")
        print(temp_cursor.fetchall())
print("n=== 示例2: 多资源管理 ===")
with MultiResourceManager() as manager:
    # 自动管理多个资源
    conn = manager.add_connection(":memory:")
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE test (id INTEGER)")
    # 可以继续添加其他资源
    # file = manager.add_file('test.txt', 'w')
    # file.write('test content')
print("n=== 示例3: 使用contextlib实用工具 ===")
from contextlib import suppress, redirect_stdout
import io
# suppress忽略指定异常
with suppress(FileNotFoundError):
    with open("不存在的文件.txt") as f:
        content = f.read()
# redirect_stdout重定向输出
output = io.StringIO()
with redirect_stdout(output):
    print("这条信息不会显示在控制台")
    print("而是被重定向到StringIO")
print(f"捕获的输出: {output.getvalue()}")

class SingletonMeta(type):
    """单例模式元类"""
    _instances = {}
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            print(f"创建 {cls.__name__} 的唯一实例")
            cls._instances[cls] = super().__call__(*args, **kwargs)
        else:
            print(f"返回 {cls.__name__} 的现有实例")
        return cls._instances[cls]
class ValidateAttributesMeta(type):
    """验证类属性的元类"""
    def __new__(mcs, name, bases, attrs):
        print(f"n创建类: {name}")
        # 验证属性名不能以下划线开头（示例规则）
        invalid_attrs = []
        for attr_name in attrs:
            if attr_name.startswith('_') and not attr_name.startswith('__'):
                invalid_attrs.append(attr_name)
        if invalid_attrs:
            print(f"警告: 属性 {invalid_attrs} 不建议以下划线开头")
        # 自动添加类版本信息
        if '__version__' not in attrs:
            attrs['__version__'] = '1.0.0'
        return super().__new__(mcs, name, bases, attrs)
# 使用元类
class Database(metaclass=SingletonMeta):
    def __init__(self, name):
        self.name = name
        self.connected = False
    def connect(self):
        self.connected = True
        print(f"{self.name} 数据库已连接")
class UserModel(metaclass=ValidateAttributesMeta):
    """用户模型类，属性会被元类验证"""
    _private_data = "不应直接访问"  # 这会触发警告
    name = "默认用户"
    email = ""
    def get_info(self):
        return f"{self.name} <{self.email}>"
# 测试
print("=== 元类示例1: 单例模式 ===")
db1 = Database("主数据库")
db1.connect()
db2 = Database("主数据库")  # 返回同一个实例
print(f"db1 is db2: {db1 is db2}")
print(f"db2.name: {db2.name}")
print("n=== 元类示例2: 属性验证 ===")
user = UserModel()
print(f"UserModel版本: {UserModel.__version__}")
print(f"用户信息: {user.get_info()}")
# 动态创建类
print("n=== 动态创建类 ===")
def class_factory(class_name, base_classes, attributes):
    """动态创建类的工厂函数"""
    return type(class_name, base_classes, attributes)
# 动态创建类
Animal = class_factory(
    'Animal',
    (),
    {
        'species': '未知',
        '__init__': lambda self, name: setattr(self, 'name', name),
        'speak': lambda self: print(f"{self.name} 发出声音")
    }
)
cat = Animal("猫咪")
cat.species = "猫科"
cat.speak()
print(f"物种: {cat.species}")

import threading
import time
import concurrent.futures
from queue import Queue, Empty
from typing import List
import random
# 线程安全的计数器
class ThreadSafeCounter:
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()
    def increment(self):
        with self._lock:  # 自动获取和释放锁
            self._value += 1
            return self._value
    @property
    def value(self):
        with self._lock:
            return self._value
# 生产者-消费者模式
class ProducerConsumer:
    def __init__(self, max_size=5):
        self.queue = Queue(maxsize=max_size)
        self.stop_event = threading.Event()
    def producer(self, producer_id: int):
        """生产者线程函数"""
        while not self.stop_event.is_set():
            item = f"产品-{producer_id}-{time.time():.2f}"
            try:
                self.queue.put(item, timeout=1)
                print(f"生产者{producer_id} 生产: {item}")
                time.sleep(random.uniform(0.1, 0.5))
            except Exception as e:
                break
    def consumer(self, consumer_id: int):
        """消费者线程函数"""
        while not self.stop_event.is_set() or not self.queue.empty():
            try:
                item = self.queue.get(timeout=1)
                print(f"消费者{consumer_id} 消费: {item}")
                self.queue.task_done()
                time.sleep(random.uniform(0.2, 0.8))
            except Empty:
                continue
    def run(self, num_producers=2, num_consumers=3, duration=5):
        """运行生产消费过程"""
        threads = []
        # 创建生产者线程
        for i in range(num_producers):
            t = threading.Thread(target=self.producer, args=(i,))
            t.daemon = True
            threads.append(t)
            t.start()
        # 创建消费者线程
        for i in range(num_consumers):
            t = threading.Thread(target=self.consumer, args=(i,))
            t.daemon = True
            threads.append(t)
            t.start()
        # 运行指定时间
        time.sleep(duration)
        self.stop_event.set()
        # 等待队列清空
        self.queue.join()
        print("所有任务完成")
# 使用ThreadPoolExecutor
def process_task(task_id: int) -> str:
    """模拟处理任务"""
    sleep_time = random.uniform(0.5, 2.0)
    print(f"任务{task_id} 开始执行，预计耗时{sleep_time:.1f}秒")
    time.sleep(sleep_time)
    result = f"任务{task_id} 完成"
    return result
def execute_with_threadpool():
    """使用线程池执行任务"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        # 提交任务
        future_to_task = {
            executor.submit(process_task, i): i 
            for i in range(10)
        }
        # 收集结果
        results = []
        for future in concurrent.futures.as_completed(future_to_task):
            task_id = future_to_task[future]
            try:
                result = future.result()
                results.append(result)
                print(result)
            except Exception as e:
                print(f"任务{task_id} 生成异常: {e}")
        print(f"n总共完成 {len(results)} 个任务")
# 测试
print("=== 线程安全计数器 ===")
counter = ThreadSafeCounter()
def worker(counter: ThreadSafeCounter, iterations: int):
    for _ in range(iterations):
        counter.increment()
threads = []
for _ in range(10):
    t = threading.Thread(target=worker, args=(counter, 100))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f"最终计数值: {counter.value} (期望值: 1000)")
print("n=== 生产者-消费者模式 ===")
pc = ProducerConsumer(max_size=3)
pc.run(duration=3)
print("n=== 线程池执行器 ===")
execute_with_threadpool()

import asyncio
import aiohttp
import asyncpg
from datetime import datetime
# 异步上下文管理器
class AsyncDatabaseConnection:
    """异步数据库连接"""
    async def __aenter__(self):
        print("异步连接数据库...")
        self.conn = await asyncpg.connect(
            user='user', password='password',
            database='test', host='localhost'
        )
        return self.conn
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("异步关闭数据库连接...")
        await self.conn.close()
# 异步迭代器
class AsyncDataStreamer:
    """异步数据流"""
    def __init__(self, limit=10, delay=0.5):
        self.limit = limit
        self.delay = delay
        self.current = 0
    def __aiter__(self):
        return self
    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(self.delay)
        data = f"数据块-{self.current}-{datetime.now().timestamp()}"
        self.current += 1
        return data
# 异步任务示例
async def fetch_url(session: aiohttp.ClientSession, url: str) -> str:
    """异步获取URL内容"""
    try:
        print(f"开始获取: {url}")
        async with session.get(url, timeout=10) as response:
            content = await response.text()
            return f"{url}: {len(content)} 字符"
    except Exception as e:
        return f"{url}: 错误 - {str(e)}"
async def process_data_stream():
    """处理异步数据流"""
    print("开始处理数据流...")
    streamer = AsyncDataStreamer(limit=5, delay=0.3)
    async for data in streamer:
        print(f"接收到: {data}")
        # 模拟数据处理
        await asyncio.sleep(0.1)
    print("数据流处理完成")
async def concurrent_url_fetcher():
    """并发获取多个URL"""
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://jsonplaceholder.typicode.com/posts/1",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print("n获取结果:")
        for result in results:
            print(f"  {result}")
async def main():
    """主异步函数"""
    print("=== 异步编程示例 ===n")
    # 示例1: 异步数据流
    await process_data_stream()
    print("n" + "="*50 + "n")
    # 示例2: 并发URL获取
    await concurrent_url_fetcher()
    print("n" + "="*50 + "n")
    # 示例3: 异步任务控制
    print("异步任务控制示例:")
    # 创建任务
    task1 = asyncio.create_task(fetch_data("任务1", 2))
    task2 = asyncio.create_task(fetch_data("任务2", 1))
    # 等待特定任务完成
    done, pending = await asyncio.wait([task1, task2], timeout=1.5)
    print(f"已完成: {len(done)} 个任务")
    print(f"仍在进行: {len(pending)} 个任务")
    # 取消剩余任务
    for task in pending:
        task.cancel()
    try:
        await asyncio.gather(*pending, return_exceptions=True)
    except asyncio.CancelledError:
        print("已取消剩余任务")
async def fetch_data(name: str, delay: float) -> str:
    """模拟异步数据获取"""
    await asyncio.sleep(delay)
    return f"{name} 完成，延迟 {delay}秒"
# 运行异步程序
if __name__ == "__main__":
    # 对于异步程序，需要特殊方式运行
    print("注意: 异步代码需要合适的运行环境")
    print("可以使用以下方式运行:")
    print("1. 在Jupyter中使用: await main()")
    print("2. 在脚本中使用: asyncio.run(main())")
    # 实际运行代码
    # asyncio.run(main())

import sys
from dataclasses import dataclass
from pympler import asizeof  # 需要安装: pip install pympler
class RegularUser:
    """普通类，使用__dict__存储属性"""
    def __init__(self, user_id, name, email, age):
        self.user_id = user_id
        self.name = name
        self.email = email
        self.age = age
class SlotUser:
    """使用__slots__优化内存的类"""
    __slots__ = ('user_id', 'name', 'email', 'age')
    def __init__(self, user_id, name, email, age):
        self.user_id = user_id
        self.name = name
        self.email = email
        self.age = age
@dataclass
class DataClassUser:
    """使用dataclass"""
    user_id: int
    name: str
    email: str
    age: int
def memory_comparison():
    """比较不同类的内存使用"""
    # 创建大量实例
    num_instances = 10000
    # 普通类
    regular_users = [RegularUser(i, f"User{i}", f"user{i}@example.com", 20+i%50) 
                     for i in range(num_instances)]
    # slots类
    slot_users = [SlotUser(i, f"User{i}", f"user{i}@example.com", 20+i%50) 
                  for i in range(num_instances)]
    # dataclass
    dataclass_users = [DataClassUser(i, f"User{i}", f"user{i}@example.com", 20+i%50) 
                       for i in range(num_instances)]
    # 计算内存使用
    reg_memory = sum(asizeof.asizeof(u) for u in regular_users[:100]) * (num_instances / 100)
    slot_memory = sum(asizeof.asizeof(u) for u in slot_users[:100]) * (num_instances / 100)
    dc_memory = sum(asizeof.asizeof(u) for u in dataclass_users[:100]) * (num_instances / 100)
    print(f"内存使用对比 ({num_instances}个实例):")
    print(f"普通类:    {reg_memory / 1024:.1f} KB")
    print(f"Slots类:   {slot_memory / 1024:.1f} KB")
    print(f"Dataclass: {dc_memory / 1024:.1f} KB")
    print(f"节省比例:  {(1 - slot_memory/reg_memory)*100:.1f}%")
# 测试
if __name__ == "__main__":
    print("=== 内存优化示例 ===")
    # 单个实例对比
    reg = RegularUser(1, "Alice", "alice@example.com", 25)
    slot = SlotUser(1, "Alice", "alice@example.com", 25)
    print(f"单个RegularUser大小: {sys.getsizeof(reg)} 字节 + __dict__")
    print(f"单个SlotUser大小: {sys.getsizeof(slot)} 字节")
    # 尝试动态添加属性
    try:
        reg.new_attr = "可以动态添加"  # 成功
        print("普通类可以动态添加属性")
    except:
        print("普通类无法动态添加属性")
    try:
        slot.new_attr = "尝试动态添加"  # 失败
        print("Slots类可以动态添加属性")
    except AttributeError:
        print("Slots类无法动态添加属性")
    print("n批量内存对比:")
    memory_comparison()

1. 循序渐进：先掌握一个主题再进入下一个，特别是异步编程需要扎实的基础
2. 动手实践：修改示例代码，观察不同参数和行为变化
3. 理解原理：了解每个特性的适用场景和优缺点，不滥用高级特性
4. 结合实际：在工作中找到应用场景，解决实际问题
5. 性能考量：只有在确实需要优化时才使用__slots__等高级优化技术

这些进阶知识点是成为Python高级开发者的关键，掌握它们能编写更高效、更优雅、更易维护的代码。建议每个主题都创建自己的练习项目，加深理解。

到此这篇关于小学生必须掌握的Python语法与代码示例的文章就介绍到这了,更多相关Python语法与代码示例内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• python语法笔记之Python raise语句用法详解
• Python核心语法与常用库完全学习指南
• Python中kwargs.get()方法语法及高级用法详解
• Python基本语法总结大全(含java、js对比)
• Python Playwright 语法知识详解(推荐)
• Python语法详解及与Java的差异对比
• 逻辑回归算法详解与Python实现完整代码示例
• Python中的循环语句详细代码示例
• Python实现数据库连接池的代码示例和最佳实践