小学生必须掌握的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

# 列表操作
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}")

# 条件与循环处理考试成绩
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 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)

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!

# 使用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()

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

# 变量命名规范
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

# 列表操作
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}")

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()