Python从入门到精通：全栈开发者的终极指南

第一部分：Python基础入门（核心语法篇）

1.1 Python简介与安装

Python是一种高级、解释型、通用编程语言，由Guido van Rossum于1991年创建。它以简洁优雅的语法和强大的功能而闻名，支持多种编程范式，包括面向对象、命令式、函数式和过程式编程。

Python的核心优势：

• 简洁易读的语法

• 丰富的标准库和第三方库

• 跨平台兼容性

• 强大的社区支持

• 在数据科学、人工智能、Web开发等领域的广泛应用

安装Python：

python

# 推荐使用Python 3.8及以上版本
# Windows：从官网下载安装包
# macOS：使用Homebrew：brew install python
# Linux：sudo apt-get install python3

# 验证安装
import sys
print(f"Python版本: {sys.version}")
print(f"Python路径: {sys.executable}")

1.2 基本语法与数据类型

变量与基本类型：

python

# 变量定义（动态类型）
name = "Python"
version = 3.9
is_awesome = True

# 数字类型
integer_num = 42           # 整数
float_num = 3.14159        # 浮点数
complex_num = 2 + 3j       # 复数

# 字符串
single_quotes = 'Hello'
double_quotes = "World"
multi_line = """这是
多行
字符串"""

# 布尔类型
true_value = True
false_value = False

# 类型检查与转换
print(type(integer_num))   # <class 'int'>
print(str(integer_num))    # "42"
print(int("100"))          # 100
print(float("3.14"))       # 3.14

数据结构：

python

# 列表 - 有序可变序列
fruits = ["apple", "banana", "cherry"]
fruits.append("orange")
fruits[0] = "avocado"

# 元组 - 有序不可变序列
coordinates = (10, 20)
# coordinates[0] = 5  # 错误！元组不可变

# 字典 - 键值对集合
person = {
    "name": "Alice",
    "age": 30,
    "city": "New York"
}
person["email"] = "alice@example.com"

# 集合 - 无序不重复元素
unique_numbers = {1, 2, 3, 3, 4}
print(unique_numbers)  # {1, 2, 3, 4}

# 字节和字节数组
byte_data = b"hello"
byte_array = bytearray(byte_data)

1.3 控制流程

条件语句：

python

# if-elif-else
age = 25

if age < 13:
    category = "儿童"
elif age < 20:
    category = "青少年"
elif age < 65:
    category = "成年人"
else:
    category = "老年人"

print(f"年龄{age}岁属于{category}")

# 三元运算符
status = "成年" if age >= 18 else "未成年"

# 模式匹配（Python 3.10+）
def http_status(code):
    match code:
        case 200:
            return "OK"
        case 404:
            return "Not Found"
        case 500:
            return "Server Error"
        case _:
            return "Unknown"

循环结构：

python

# for循环
for i in range(5):          # 0到4
    print(f"迭代 {i}")

for fruit in ["apple", "banana", "cherry"]:
    print(f"我喜欢吃{fruit}")

# 使用enumerate获取索引
for index, fruit in enumerate(["apple", "banana"]):
    print(f"{index}: {fruit}")

# while循环
count = 0
while count < 5:
    print(f"计数: {count}")
    count += 1

# 循环控制语句
for i in range(10):
    if i == 3:
        continue  # 跳过当前迭代
    if i == 7:
        break     # 终止循环
    print(i)
else:
    print("循环正常结束")  # 仅在循环未break时执行

# 列表推导式
squares = [x**2 for x in range(10)]
even_squares = [x**2 for x in range(10) if x % 2 == 0]

1.4 函数基础

函数定义与调用：

python

# 基本函数
def greet(name):
    """返回问候语"""
    return f"Hello, {name}!"

# 调用函数
print(greet("Alice"))

# 默认参数
def power(base, exponent=2):
    return base ** exponent

print(power(3))      # 9
print(power(3, 3))   # 27

# 可变参数
def sum_all(*args):
    """计算任意数量数字的和"""
    return sum(args)

print(sum_all(1, 2, 3, 4, 5))  # 15

# 关键字参数
def create_profile(**kwargs):
    for key, value in kwargs.items():
        print(f"{key}: {value}")

create_profile(name="Bob", age=25, city="Tokyo")

# 混合参数
def complex_function(a, b, *args, option=True, **kwargs):
    print(f"a={a}, b={b}, args={args}, option={option}, kwargs={kwargs}")

# 类型提示（Python 3.5+）
def add_numbers(a: int, b: int) -> int:
    return a + b

# Lambda表达式
square = lambda x: x ** 2
print(square(5))  # 25

# 排序中使用lambda
pairs = [(1, 'one'), (2, 'two'), (3, 'three')]
pairs.sort(key=lambda pair: pair[1])

第二部分：Python进阶编程

2.1 面向对象编程（OOP）

类与对象：

python

class Person:
    # 类属性
    species = "Homo sapiens"
    
    # 初始化方法
    def __init__(self, name, age):
        # 实例属性
        self.name = name
        self.age = age
        self.__private_attr = "secret"  # 私有属性
    
    # 实例方法
    def introduce(self):
        return f"我叫{self.name}，今年{self.age}岁"
    
    # 类方法
    @classmethod
    def create_baby(cls, name):
        return cls(name, 0)
    
    # 静态方法
    @staticmethod
    def is_adult(age):
        return age >= 18
    
    # 属性装饰器
    @property
    def birth_year(self):
        from datetime import datetime
        return datetime.now().year - self.age
    
    @birth_year.setter
    def birth_year(self, year):
        from datetime import datetime
        self.age = datetime.now().year - year
    
    # 特殊方法
    def __str__(self):
        return f"Person(name={self.name}, age={self.age})"
    
    def __repr__(self):
        return f"Person('{self.name}', {self.age})"
    
    def __add__(self, other):
        if isinstance(other, Person):
            return Family([self, other])
        raise TypeError("只能与Person对象相加")

# 继承
class Student(Person):
    def __init__(self, name, age, student_id):
        super().__init__(name, age)
        self.student_id = student_id
    
    def introduce(self):
        base_intro = super().introduce()
        return f"{base_intro}，学号是{self.student_id}"

# 多态
class Teacher(Person):
    def introduce(self):
        return f"我是{self.name}老师，任教{self.age}年"

# 使用示例
alice = Person("Alice", 25)
print(alice.introduce())
print(alice.birth_year)

baby = Person.create_baby("Baby")
print(baby)

# 多重继承与Mixin
class JSONSerializableMixin:
    def to_json(self):
        import json
        return json.dumps(self.__dict__)

class Employee(Person, JSONSerializableMixin):
    def __init__(self, name, age, employee_id):
        super().__init__(name, age)
        self.employee_id = employee_id

emp = Employee("Bob", 30, "E123")
print(emp.to_json())

2.2 函数式编程

高阶函数与装饰器：

python

# 高阶函数：接受或返回函数的函数
def apply_twice(func, value):
    return func(func(value))

def add_five(x):
    return x + 5

print(apply_twice(add_five, 10))  # 20

# 内置高阶函数
numbers = [1, 2, 3, 4, 5]

# map
squared = list(map(lambda x: x**2, numbers))

# filter
evens = list(filter(lambda x: x % 2 == 0, numbers))

# reduce
from functools import reduce
product = reduce(lambda x, y: x * y, numbers)

# 装饰器
def timer_decorator(func):
    import time
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__}执行时间: {end_time - start_time:.4f}秒")
        return result
    return wrapper

def cache_decorator(func):
    cache = {}
    def wrapper(*args):
        if args in cache:
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return wrapper

@timer_decorator
@cache_decorator
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(10))

# 带参数的装饰器
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(3)
def say_hello(name):
    print(f"Hello, {name}!")

say_hello("World")

生成器与迭代器：

python

# 生成器函数
def countdown(n):
    while n > 0:
        yield n
        n -= 1

for number in countdown(5):
    print(number)  # 5, 4, 3, 2, 1

# 生成器表达式
squares_gen = (x**2 for x in range(10))
print(sum(squares_gen))  # 285

# 自定义迭代器
class FibonacciIterator:
    def __init__(self, max_count):
        self.max_count = max_count
        self.count = 0
        self.a, self.b = 0, 1
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.count >= self.max_count:
            raise StopIteration
        value = self.a
        self.a, self.b = self.b, self.a + self.b
        self.count += 1
        return value

fib_iter = FibonacciIterator(10)
for num in fib_iter:
    print(num)

2.3 异常处理与调试

异常处理：

python

try:
    # 可能引发异常的代码
    numerator = 10
    denominator = 0
    result = numerator / denominator
    
except ZeroDivisionError as e:
    print(f"除零错误: {e}")
    result = None
    
except (TypeError, ValueError) as e:
    print(f"类型或值错误: {e}")
    
except Exception as e:
    print(f"未知错误: {e}")
    
else:
    print("没有发生异常")
    
finally:
    print("无论是否异常都会执行")

# 自定义异常
class InsufficientFundsError(Exception):
    def __init__(self, balance, amount):
        super().__init__(f"余额不足: 当前余额{balance}，需要{amount}")
        self.balance = balance
        self.amount = amount

class BankAccount:
    def __init__(self, balance):
        self.balance = balance
    
    def withdraw(self, amount):
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)
        self.balance -= amount
        return self.balance

# 上下文管理器
class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name
    
    def __enter__(self):
        print(f"连接数据库: {self.db_name}")
        self.connection = {"connected": True}
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"关闭数据库连接: {self.db_name}")
        if exc_type:
            print(f"发生异常: {exc_type.__name__}: {exc_val}")
        return False  # 不抑制异常

with DatabaseConnection("my_db") as conn:
    print(f"连接状态: {conn}")
    # 执行数据库操作

2.4 模块与包管理

创建模块和包：

text

# 项目结构示例：
my_package/
├── __init__.py
├── module1.py
├── module2.py
└── subpackage/
    ├── __init__.py
    └── module3.py

python

# module1.py
def function1():
    return "来自module1"

class Class1:
    pass

# __init__.py
from .module1 import function1
from .module2 import Class2

__all__ = ['function1', 'Class2']  # 控制from package import *的行为

# 使用包
import my_package
from my_package import function1
from my_package.subpackage import module3

# 相对导入
# 在subpackage/module3.py中
from ..module1 import function1

包管理与虚拟环境：

bash

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows: venv\Scripts\activate
# macOS/Linux: source venv/bin/activate

# 安装包
pip install numpy pandas flask

# 从requirements.txt安装
pip install -r requirements.txt

# 生成requirements.txt
pip freeze > requirements.txt

# 使用pipenv（更现代的包管理工具）
pip install pipenv
pipenv install numpy
pipenv shell

第三部分：Python核心应用领域

3.1 数据科学与分析

NumPy数组计算：

python

import numpy as np

# 创建数组
arr1 = np.array([1, 2, 3, 4, 5])
arr2 = np.arange(0, 10, 2)  # [0, 2, 4, 6, 8]
arr3 = np.linspace(0, 1, 5)  # 均匀分布 [0., 0.25, 0.5, 0.75, 1.]

# 多维数组
matrix = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
zeros = np.zeros((3, 4))  # 3x4零矩阵
ones = np.ones((2, 3))    # 2x3单位矩阵
identity = np.eye(3)      # 3x3单位矩阵

# 数组操作
reshaped = arr1.reshape(5, 1)  # 改变形状
flattened = matrix.flatten()   # 展平为一维

# 数学运算
sum_arr = np.sum(arr1)
mean_arr = np.mean(arr1)
std_arr = np.std(arr1)

# 矩阵运算
dot_product = np.dot(matrix, matrix.T)  # 矩阵乘法
determinant = np.linalg.det(matrix)     # 行列式
inverse = np.linalg.inv(matrix)         # 逆矩阵

# 广播机制
arr = np.array([1, 2, 3])
result = arr * 2  # [2, 4, 6]

Pandas数据处理：

python

import pandas as pd
import numpy as np

# 创建Series
s = pd.Series([1, 3, 5, np.nan, 6, 8])

# 创建DataFrame
dates = pd.date_range('20230101', periods=6)
df = pd.DataFrame(np.random.randn(6, 4), 
                  index=dates, 
                  columns=['A', 'B', 'C', 'D'])

# 数据查看
print(df.head())      # 前5行
print(df.tail(3))     # 后3行
print(df.describe())  # 统计摘要
print(df.info())      # 信息概览

# 选择数据
print(df['A'])           # 选择列
print(df[0:3])           # 选择行
print(df.loc['20230102']) # 按标签选择
print(df.iloc[3])        # 按位置选择

# 条件筛选
filtered = df[df['A'] > 0]  # A列大于0的行

# 数据清洗
df_filled = df.fillna(0)          # 填充缺失值
df_dropped = df.dropna()          # 删除缺失值
df_no_duplicates = df.drop_duplicates()  # 删除重复值

# 数据分组与聚合
grouped = df.groupby(lambda x: x.month).mean()  # 按月分组求平均

# 合并数据
df1 = pd.DataFrame({'A': ['A0', 'A1'], 'B': ['B0', 'B1']})
df2 = pd.DataFrame({'A': ['A2', 'A3'], 'B': ['B2', 'B3']})
result = pd.concat([df1, df2])  # 纵向合并

# 读写数据
df.to_csv('data.csv')          # 保存为CSV
df_read = pd.read_csv('data.csv')  # 读取CSV
df.to_excel('data.xlsx')       # 保存为Excel
df_read_excel = pd.read_excel('data.xlsx')  # 读取Excel

数据可视化：

python

import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np

# 设置样式
plt.style.use('seaborn-v0_8-darkgrid')

# 创建数据
x = np.linspace(0, 10, 100)
y1 = np.sin(x)
y2 = np.cos(x)

# 创建图形和坐标轴
fig, axes = plt.subplots(2, 2, figsize=(12, 8))

# 折线图
axes[0, 0].plot(x, y1, label='sin(x)', color='blue', linewidth=2)
axes[0, 0].plot(x, y2, label='cos(x)', color='red', linestyle='--')
axes[0, 0].set_title('三角函数')
axes[0, 0].set_xlabel('x')
axes[0, 0].set_ylabel('y')
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)

# 散点图
np.random.seed(42)
x_scatter = np.random.randn(100)
y_scatter = np.random.randn(100)
colors = np.random.rand(100)
sizes = 1000 * np.random.rand(100)

scatter = axes[0, 1].scatter(x_scatter, y_scatter, 
                            c=colors, s=sizes, alpha=0.6, 
                            cmap='viridis')
axes[0, 1].set_title('散点图')
plt.colorbar(scatter, ax=axes[0, 1])

# 柱状图
categories = ['A', 'B', 'C', 'D', 'E']
values = [23, 45, 56, 78, 33]
bars = axes[1, 0].bar(categories, values, 
                      color=['red', 'blue', 'green', 'orange', 'purple'])
axes[1, 0].set_title('柱状图')
axes[1, 0].set_xlabel('类别')
axes[1, 0].set_ylabel('值')

# 添加数值标签
for bar in bars:
    height = bar.get_height()
    axes[1, 0].text(bar.get_x() + bar.get_width()/2., height,
                   f'{int(height)}', ha='center', va='bottom')

# 直方图
data_hist = np.random.randn(1000)
axes[1, 1].hist(data_hist, bins=30, edgecolor='black', alpha=0.7)
axes[1, 1].set_title('数据分布直方图')
axes[1, 1].set_xlabel('值')
axes[1, 1].set_ylabel('频数')

# 调整布局
plt.tight_layout()
plt.show()

# 使用Seaborn绘制更复杂的统计图形
import seaborn as sns

# 加载示例数据集
tips = sns.load_dataset('tips')

# 创建多面板图形
fig, axes = plt.subplots(2, 2, figsize=(12, 10))

# 箱线图
sns.boxplot(x='day', y='total_bill', data=tips, ax=axes[0, 0])
axes[0, 0].set_title('每日消费箱线图')

# 小提琴图
sns.violinplot(x='day', y='total_bill', hue='sex', 
               data=tips, split=True, ax=axes[0, 1])
axes[0, 1].set_title('性别与消费小提琴图')

# 热力图
correlation = tips.corr()
sns.heatmap(correlation, annot=True, cmap='coolwarm', 
            ax=axes[1, 0])
axes[1, 0].set_title('相关性热力图')

# 联合分布图
sns.jointplot(x='total_bill', y='tip', data=tips, 
              kind='reg', color='green')
plt.suptitle('消费与小费联合分布', y=1.02)

plt.tight_layout()
plt.show()

3.2 Web开发与API

Flask Web框架：

python

from flask import Flask, render_template, request, jsonify, redirect, url_for, session
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required
import os

app = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(24)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'

# 数据模型
class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password = db.Column(db.String(60), nullable=False)
    posts = db.relationship('Post', backref='author', lazy=True)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

# 路由
@app.route('/')
def home():
    posts = Post.query.all()
    return render_template('home.html', posts=posts)

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form['username']
        email = request.form['email']
        password = request.form['password']
        
        user = User(username=username, email=email, password=password)
        db.session.add(user)
        db.session.commit()
        
        return redirect(url_for('login'))
    
    return render_template('register.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        email = request.form['email']
        password = request.form['password']
        
        user = User.query.filter_by(email=email).first()
        if user and user.password == password:
            login_user(user)
            return redirect(url_for('dashboard'))
    
    return render_template('login.html')

@app.route('/dashboard')
@login_required
def dashboard():
    return render_template('dashboard.html')

@app.route('/api/posts', methods=['GET'])
def get_posts():
    posts = Post.query.all()
    return jsonify([{
        'id': post.id,
        'title': post.title,
        'content': post.content
    } for post in posts])

@app.route('/api/posts', methods=['POST'])
def create_post():
    data = request.get_json()
    post = Post(title=data['title'], content=data['content'])
    db.session.add(post)
    db.session.commit()
    return jsonify({'message': 'Post created!'}), 201

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

FastAPI高性能API：

python

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, EmailStr
from datetime import datetime, timedelta
from typing import Optional, List
from jose import JWTError, jwt
from passlib.context import CryptContext
import databases
import sqlalchemy
from sqlalchemy import create_engine

# 数据库配置
DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()

# 定义数据表
users = sqlalchemy.Table(
    "users",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("username", sqlalchemy.String, unique=True),
    sqlalchemy.Column("email", sqlalchemy.String, unique=True),
    sqlalchemy.Column("hashed_password", sqlalchemy.String),
    sqlalchemy.Column("is_active", sqlalchemy.Boolean, default=True),
)

# Pydantic模型
class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    
    class Config:
        orm_mode = True

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

# 安全配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI(title="用户管理系统", 
              description="完整的用户认证和管理API",
              version="1.0.0")

# 工具函数
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的认证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    
    query = users.select().where(users.c.username == token_data.username)
    user = await database.fetch_one(query)
    if user is None:
        raise credentials_exception
    return User(**user)

# 事件处理器
@app.on_event("startup")
async def startup():
    await database.connect()
    engine = create_engine(DATABASE_URL)
    metadata.create_all(engine)

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

# API端点
@app.post("/register", response_model=User, status_code=201)
async def register(user: UserCreate):
    # 检查用户是否存在
    query = users.select().where(users.c.username == user.username)
    existing_user = await database.fetch_one(query)
    if existing_user:
        raise HTTPException(status_code=400, detail="用户名已存在")
    
    # 创建用户
    hashed_password = get_password_hash(user.password)
    query = users.insert().values(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    user_id = await database.execute(query)
    
    # 返回创建的用户
    query = users.select().where(users.c.id == user_id)
    created_user = await database.fetch_one(query)
    return User(**created_user)

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # 验证用户
    query = users.select().where(users.c.username == form_data.username)
    user = await database.fetch_one(query)
    
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 生成token
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/users", response_model=List[User])
async def read_users(skip: int = 0, limit: int = 100):
    query = users.select().offset(skip).limit(limit)
    return await database.fetch_all(query)

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3.3 机器学习与深度学习

Scikit-learn机器学习：

python

import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
import seaborn as sns

# 加载数据
from sklearn.datasets import load_iris
iris = load_iris()
X = iris.data
y = iris.target

# 数据预处理
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

# 特征标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 创建多个分类器进行比较
classifiers = {
    'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42),
    'SVM': SVC(kernel='rbf', probability=True, random_state=42),
    'Gradient Boosting': GradientBoostingClassifier(n_estimators=100, random_state=42)
}

# 训练和评估模型
results = {}
for name, clf in classifiers.items():
    # 交叉验证
    cv_scores = cross_val_score(clf, X_train_scaled, y_train, cv=5)
    
    # 训练模型
    clf.fit(X_train_scaled, y_train)
    
    # 预测
    y_pred = clf.predict(X_test_scaled)
    
    # 评估
    accuracy = accuracy_score(y_test, y_pred)
    report = classification_report(y_test, y_pred, target_names=iris.target_names, output_dict=True)
    
    results[name] = {
        'cv_mean': cv_scores.mean(),
        'cv_std': cv_scores.std(),
        'test_accuracy': accuracy,
        'classification_report': report
    }
    
    print(f"{name}:")
    print(f"  交叉验证准确率: {cv_scores.mean():.3f} (+/- {cv_scores.std() * 2:.3f})")
    print(f"  测试集准确率: {accuracy:.3f}")
    print()

# 可视化结果
fig, axes = plt.subplots(1, 3, figsize=(15, 4))

# 准确率比较
accuracies = [results[name]['test_accuracy'] for name in classifiers.keys()]
axes[0].bar(classifiers.keys(), accuracies)
axes[0].set_title('模型准确率比较')
axes[0].set_ylabel('准确率')
axes[0].set_ylim(0, 1)

# 特征重要性（随机森林）
rf = classifiers['Random Forest']
importances = rf.feature_importances_
indices = np.argsort(importances)[::-1]
features = iris.feature_names

axes[1].bar(range(X.shape[1]), importances[indices])
axes[1].set_title('特征重要性（随机森林）')
axes[1].set_xticks(range(X.shape[1]))
axes[1].set_xticklabels([features[i] for i in indices], rotation=45)

# 混淆矩阵
best_model_name = max(results, key=lambda x: results[x]['test_accuracy'])
best_model = classifiers[best_model_name]
y_pred_best = best_model.predict(X_test_scaled)
cm = confusion_matrix(y_test, y_pred_best)

sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
            xticklabels=iris.target_names, 
            yticklabels=iris.target_names, ax=axes[2])
axes[2].set_title(f'{best_model_name}混淆矩阵')
axes[2].set_xlabel('预测标签')
axes[2].set_ylabel('真实标签')

plt.tight_layout()
plt.show()

# 超参数调优示例
param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [None, 10, 20, 30],
    'min_samples_split': [2, 5, 10],
    'min_samples_leaf': [1, 2, 4]
}

rf = RandomForestClassifier(random_state=42)
grid_search = GridSearchCV(rf, param_grid, cv=5, scoring='accuracy', n_jobs=-1)
grid_search.fit(X_train_scaled, y_train)

print(f"最佳参数: {grid_search.best_params_}")
print(f"最佳交叉验证分数: {grid_search.best_score_:.3f}")

# 使用PCA进行降维可视化
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)

plt.figure(figsize=(10, 8))
colors = ['red', 'green', 'blue']
for i, color in enumerate(colors):
    plt.scatter(X_pca[y == i, 0], X_pca[y == i, 1], 
                color=color, alpha=0.8, 
                label=iris.target_names[i])
plt.xlabel('第一主成分')
plt.ylabel('第二主成分')
plt.title('PCA降维可视化')
plt.legend()
plt.show()

TensorFlow/Keras深度学习：

python

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers, models, datasets, callbacks
import numpy as np
import matplotlib.pyplot as plt

# 加载和准备数据
(x_train, y_train), (x_test, y_test) = datasets.cifar10.load_data()

# 数据标准化
x_train = x_train.astype('float32') / 255.0
x_test = x_test.astype('float32') / 255.0

# 将标签转换为one-hot编码
y_train = keras.utils.to_categorical(y_train, 10)
y_test = keras.utils.to_categorical(y_test, 10)

# 数据增强
data_augmentation = keras.Sequential([
    layers.RandomFlip("horizontal"),
    layers.RandomRotation(0.1),
    layers.RandomZoom(0.1),
])

# 构建卷积神经网络
def create_cnn_model(input_shape=(32, 32, 3), num_classes=10):
    inputs = keras.Input(shape=input_shape)
    
    # 数据增强层
    x = data_augmentation(inputs)
    
    # 第一个卷积块
    x = layers.Conv2D(32, (3, 3), activation='relu', padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Conv2D(32, (3, 3), activation='relu', padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)
    x = layers.Dropout(0.25)(x)
    
    # 第二个卷积块
    x = layers.Conv2D(64, (3, 3), activation='relu', padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Conv2D(64, (3, 3), activation='relu', padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)
    x = layers.Dropout(0.25)(x)
    
    # 第三个卷积块
    x = layers.Conv2D(128, (3, 3), activation='relu', padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Conv2D(128, (3, 3), activation='relu', padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)
    x = layers.Dropout(0.25)(x)
    
    # 全连接层
    x = layers.Flatten()(x)
    x = layers.Dense(256, activation='relu')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Dropout(0.5)(x)
    x = layers.Dense(128, activation='relu')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Dropout(0.5)(x)
    
    # 输出层
    outputs = layers.Dense(num_classes, activation='softmax')(x)
    
    model = keras.Model(inputs=inputs, outputs=outputs)
    return model

# 创建模型
model = create_cnn_model()

# 编译模型
model.compile(
    optimizer=keras.optimizers.Adam(learning_rate=0.001),
    loss='categorical_crossentropy',
    metrics=['accuracy']
)

# 定义回调函数
callbacks_list = [
    callbacks.EarlyStopping(
        monitor='val_accuracy',
        patience=10,
        restore_best_weights=True
    ),
    callbacks.ReduceLROnPlateau(
        monitor='val_loss',
        factor=0.5,
        patience=5,
        min_lr=1e-6
    ),
    callbacks.ModelCheckpoint(
        'best_model.h5',
        monitor='val_accuracy',
        save_best_only=True
    )
]

# 训练模型
history = model.fit(
    x_train, y_train,
    batch_size=64,
    epochs=50,
    validation_split=0.2,
    callbacks=callbacks_list,
    verbose=1
)

# 评估模型
test_loss, test_acc = model.evaluate(x_test, y_test, verbose=0)
print(f"测试准确率: {test_acc:.4f}")
print(f"测试损失: {test_loss:.4f}")

# 可视化训练过程
fig, axes = plt.subplots(1, 2, figsize=(12, 4))

# 准确率曲线
axes[0].plot(history.history['accuracy'], label='训练准确率')
axes[0].plot(history.history['val_accuracy'], label='验证准确率')
axes[0].set_title('模型准确率')
axes[0].set_xlabel('Epoch')
axes[0].set_ylabel('准确率')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 损失曲线
axes[1].plot(history.history['loss'], label='训练损失')
axes[1].plot(history.history['val_loss'], label='验证损失')
axes[1].set_title('模型损失')
axes[1].set_xlabel('Epoch')
axes[1].set_ylabel('损失')
axes[1].legend()
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

# 预测示例
class_names = ['飞机', '汽车', '鸟', '猫', '鹿', 
               '狗', '青蛙', '马', '船', '卡车']

# 随机选择一些测试图像
num_images = 10
random_indices = np.random.choice(len(x_test), num_images, replace=False)
sample_images = x_test[random_indices]
sample_labels = np.argmax(y_test[random_indices], axis=1)

# 进行预测
predictions = model.predict(sample_images)
predicted_classes = np.argmax(predictions, axis=1)

# 可视化预测结果
plt.figure(figsize=(15, 8))
for i in range(num_images):
    plt.subplot(2, 5, i + 1)
    plt.imshow(sample_images[i])
    
    # 设置标题颜色（绿色表示正确，红色表示错误）
    color = 'green' if predicted_classes[i] == sample_labels[i] else 'red'
    
    plt.title(f"真实: {class_names[sample_labels[i]]}\n预测: {class_names[predicted_classes[i]]}", 
              color=color)
    plt.axis('off')

plt.suptitle('模型预测结果示例', fontsize=16)
plt.tight_layout()
plt.show()

# 保存模型
model.save('cifar10_cnn_model.h5')

# 加载模型
loaded_model = keras.models.load_model('cifar10_cnn_model.h5')

# 转换为TensorFlow Lite格式（用于移动设备）
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()

with open('model.tflite', 'wb') as f:
    f.write(tflite_model)

3.4 自动化与脚本编程

文件与系统操作：

python

import os
import shutil
import glob
import sys
import subprocess
from pathlib import Path
import json
import csv
import pickle
import hashlib
import zipfile
import tarfile
from datetime import datetime
import time

# 文件操作
def file_operations_demo():
    # 创建目录
    os.makedirs("test_dir/subdir", exist_ok=True)
    
    # 写入文件
    with open("test_dir/test.txt", "w") as f:
        f.write("Hello, World!\n")
        f.write("This is a test file.\n")
    
    # 读取文件
    with open("test_dir/test.txt", "r") as f:
        content = f.read()
        print("文件内容:")
        print(content)
    
    # 追加内容
    with open("test_dir/test.txt", "a") as f:
        f.write("Appended content.\n")
    
    # 读取行
    with open("test_dir/test.txt", "r") as f:
        lines = f.readlines()
        print("\n按行读取:")
        for i, line in enumerate(lines, 1):
            print(f"{i}: {line.strip()}")
    
    # 使用Path对象（推荐）
    path = Path("test_dir/test.txt")
    print(f"\n文件路径: {path}")
    print(f"文件是否存在: {path.exists()}")
    print(f"文件大小: {path.stat().st_size} 字节")
    print(f"最后修改时间: {datetime.fromtimestamp(path.stat().st_mtime)}")
    
    # 遍历目录
    print("\n目录内容:")
    for item in Path(".").iterdir():
        if item.is_dir():
            print(f"目录: {item}")
        elif item.is_file():
            print(f"文件: {item}")

# CSV文件处理
def csv_operations():
    # 写入CSV
    with open("data.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["姓名", "年龄", "城市"])
        writer.writerow(["张三", 25, "北京"])
        writer.writerow(["李四", 30, "上海"])
        writer.writerow(["王五", 28, "广州"])
    
    # 读取CSV
    with open("data.csv", "r", encoding="utf-8") as f:
        reader = csv.reader(f)
        print("\nCSV内容:")
        for row in reader:
            print(row)
    
    # 使用DictReader
    with open("data.csv", "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        print("\n使用DictReader:")
        for row in reader:
            print(f"{row['姓名']}, {row['年龄']}岁, 来自{row['城市']}")

# JSON处理
def json_operations():
    data = {
        "name": "张三",
        "age": 30,
        "city": "北京",
        "hobbies": ["阅读", "游泳", "编程"],
        "education": {
            "degree": "硕士",
            "school": "清华大学"
        }
    }
    
    # 写入JSON
    with open("data.json", "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    
    # 读取JSON
    with open("data.json", "r", encoding="utf-8") as f:
        loaded_data = json.load(f)
        print("\nJSON数据:")
        print(json.dumps(loaded_data, ensure_ascii=False, indent=2))

# 系统命令执行
def system_commands():
    # 执行简单命令
    result = subprocess.run(["ls", "-la"], capture_output=True, text=True)
    print("目录列表:")
    print(result.stdout)
    
    # 获取系统信息
    print(f"Python版本: {sys.version}")
    print(f"操作系统: {sys.platform}")
    print(f"当前工作目录: {os.getcwd()}")
    
    # 环境变量
    print(f"\nPATH环境变量:")
    print(os.environ.get("PATH", "").split(":")[:5])

# 文件哈希和校验
def file_hash_demo():
    filename = "test_dir/test.txt"
    
    # 计算MD5哈希
    with open(filename, "rb") as f:
        md5_hash = hashlib.md5()
        for chunk in iter(lambda: f.read(4096), b""):
            md5_hash.update(chunk)
        print(f"MD5哈希: {md5_hash.hexdigest()}")
    
    # 计算SHA256哈希
    with open(filename, "rb") as f:
        sha256_hash = hashlib.sha256()
        for chunk in iter(lambda: f.read(4096), b""):
            sha256_hash.update(chunk)
        print(f"SHA256哈希: {sha256_hash.hexdigest()}")

# 压缩文件处理
def compression_demo():
    # 创建ZIP文件
    with zipfile.ZipFile("archive.zip", "w") as zipf:
        zipf.write("test_dir/test.txt", arcname="test.txt")
        zipf.write("data.csv", arcname="data.csv")
    
    # 读取ZIP文件
    with zipfile.ZipFile("archive.zip", "r") as zipf:
        print("ZIP文件内容:")
        for file_info in zipf.infolist():
            print(f"{file_info.filename} - {file_info.file_size} 字节")
    
    # 解压ZIP文件
    with zipfile.ZipFile("archive.zip", "r") as zipf:
        zipf.extractall("extracted")

# 定时任务和计划
def scheduled_tasks():
    import schedule
    import time
    
    def job():
        print(f"定时任务执行: {datetime.now()}")
    
    def another_job():
        print(f"另一个任务: {datetime.now()}")
    
    # 安排任务
    schedule.every(10).seconds.do(job)
    schedule.every().minute.do(another_job)
    schedule.every().day.at("10:30").do(job)
    
    print("开始定时任务调度...")
    while True:
        schedule.run_pending()
        time.sleep(1)

# 主函数
if __name__ == "__main__":
    print("=== 文件与系统操作演示 ===\n")
    
    file_operations_demo()
    csv_operations()
    json_operations()
    system_commands()
    file_hash_demo()
    compression_demo()
    
    # 清理临时文件
    print("\n=== 清理临时文件 ===")
    if os.path.exists("test_dir"):
        shutil.rmtree("test_dir")
    for file in ["data.csv", "data.json", "archive.zip"]:
        if os.path.exists(file):
            os.remove(file)
    if os.path.exists("extracted"):
        shutil.rmtree("extracted")
    
    print("演示完成！")

第四部分：Python高级主题与最佳实践

4.1 并发与并行编程

python

import asyncio
import concurrent.futures
import threading
import multiprocessing
import time
from typing import List
import aiohttp

# 多线程
def thread_demo():
    def worker(number):
        print(f"线程 {number} 开始")
        time.sleep(2)
        print(f"线程 {number} 结束")
        return number * number
    
    threads = []
    results = []
    
    for i in range(5):
        thread = threading.Thread(target=worker, args=(i,))
        threads.append(thread)
        thread.start()
    
    for thread in threads:
        thread.join()
    
    print("所有线程执行完成")

# 线程池
def thread_pool_demo():
    def task(n):
        time.sleep(1)
        return n * n
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        # 提交任务
        futures = [executor.submit(task, i) for i in range(10)]
        
        # 获取结果
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            print(f"任务结果: {result}")

# 多进程
def process_demo():
    def cpu_intensive_task(n):
        return sum(i * i for i in range(n))
    
    numbers = [1000000, 2000000, 3000000, 4000000]
    
    # 顺序执行
    start_time = time.time()
    results = [cpu_intensive_task(n) for n in numbers]
    seq_time = time.time() - start_time
    print(f"顺序执行时间: {seq_time:.2f}秒")
    
    # 多进程执行
    start_time = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = list(executor.map(cpu_intensive_task, numbers))
    parallel_time = time.time() - start_time
    print(f"并行执行时间: {parallel_time:.2f}秒")
    print(f"加速比: {seq_time / parallel_time:.2f}")

# 异步编程
async def async_demo():
    async def fetch_url(url):
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()
    
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3"
    ]
    
    print("开始异步请求...")
    start_time = time.time()
    
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    elapsed_time = time.time() - start_time
    print(f"总耗时: {elapsed_time:.2f}秒")
    print(f"响应数量: {len(results)}")

# 协程生成器
async def coroutine_generator():
    async def producer(queue: asyncio.Queue):
        for i in range(10):
            await asyncio.sleep(0.5)
            await queue.put(i)
            print(f"生产: {i}")
        await queue.put(None)  # 结束信号
    
    async def consumer(queue: asyncio.Queue):
        while True:
            item = await queue.get()
            if item is None:
                break
            await asyncio.sleep(1)
            print(f"消费: {item}")
            queue.task_done()
    
    queue = asyncio.Queue(maxsize=5)
    
    producer_task = asyncio.create_task(producer(queue))
    consumer_task = asyncio.create_task(consumer(queue))
    
    await asyncio.gather(producer_task, consumer_task)

# 性能比较
def performance_comparison():
    def io_bound_task(n):
        time.sleep(0.1)  # 模拟IO操作
        return n * n
    
    def cpu_bound_task(n):
        return sum(i * i for i in range(n))
    
    tasks = list(range(100))
    
    # IO密集型任务 - 多线程优势
    print("IO密集型任务测试:")
    
    start_time = time.time()
    results = [io_bound_task(i) for i in tasks]
    print(f"顺序执行: {time.time() - start_time:.2f}秒")
    
    start_time = time.time()
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(io_bound_task, tasks))
    print(f"线程池执行: {time.time() - start_time:.2f}秒")
    
    # CPU密集型任务 - 多进程优势
    print("\nCPU密集型任务测试:")
    
    start_time = time.time()
    results = [cpu_bound_task(10000) for _ in range(10)]
    print(f"顺序执行: {time.time() - start_time:.2f}秒")
    
    start_time = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = list(executor.map(lambda x: cpu_bound_task(10000), range(10)))
    print(f"进程池执行: {time.time() - start_time:.2f}秒")

if __name__ == "__main__":
    print("=== 并发与并行编程演示 ===\n")
    
    print("1. 多线程演示:")
    thread_demo()
    
    print("\n2. 线程池演示:")
    thread_pool_demo()
    
    print("\n3. 多进程演示:")
    process_demo()
    
    print("\n4. 异步编程演示:")
    asyncio.run(async_demo())
    
    print("\n5. 协程生成器模式:")
    asyncio.run(coroutine_generator())
    
    print("\n6. 性能比较:")
    performance_comparison()

4.2 性能优化与调试

python

import time
import cProfile
import pstats
import tracemalloc
import line_profiler
import memory_profiler
from functools import lru_cache
import numpy as np
from numba import jit, njit, vectorize
import cython
import gc

# 1. 算法优化
def algorithm_optimization():
    # 低效实现
    def inefficient_find_duplicates(arr):
        duplicates = []
        for i in range(len(arr)):
            for j in range(i + 1, len(arr)):
                if arr[i] == arr[j] and arr[i] not in duplicates:
                    duplicates.append(arr[i])
        return duplicates
    
    # 高效实现
    def efficient_find_duplicates(arr):
        seen = set()
        duplicates = set()
        for item in arr:
            if item in seen:
                duplicates.add(item)
            else:
                seen.add(item)
        return list(duplicates)
    
    # 测试
    test_array = list(range(1000)) + [999] * 10
    
    start_time = time.time()
    result1 = inefficient_find_duplicates(test_array)
    time1 = time.time() - start_time
    
    start_time = time.time()
    result2 = efficient_find_duplicates(test_array)
    time2 = time.time() - start_time
    
    print(f"低效算法时间: {time1:.4f}秒")
    print(f"高效算法时间: {time2:.4f}秒")
    print(f"加速比: {time1 / time2:.2f}倍")

# 2. 缓存优化
def cache_optimization():
    @lru_cache(maxsize=128)
    def fibonacci_cached(n):
        if n < 2:
            return n
        return fibonacci_cached(n-1) + fibonacci_cached(n-2)
    
    def fibonacci_uncached(n):
        if n < 2:
            return n
        return fibonacci_uncached(n-1) + fibonacci_uncached(n-2)
    
    print("斐波那契数列计算对比:")
    
    # 测试缓存效果
    start_time = time.time()
    for i in range(30, 35):
        fibonacci_cached(i)
    cached_time = time.time() - start_time
    
    start_time = time.time()
    for i in range(30, 35):
        fibonacci_uncached(i)
    uncached_time = time.time() - start_time
    
    print(f"使用缓存时间: {cached_time:.4f}秒")
    print(f"未使用缓存时间: {uncached_time:.4f}秒")
    print(f"缓存加速比: {uncached_time / cached_time:.2f}倍")

# 3. 使用NumPy向量化
def numpy_vectorization():
    size = 1000000
    
    # Python循环
    start_time = time.time()
    python_list = list(range(size))
    python_squares = [x**2 for x in python_list]
    python_time = time.time() - start_time
    
    # NumPy向量化
    start_time = time.time()
    numpy_array = np.arange(size)
    numpy_squares = numpy_array ** 2
    numpy_time = time.time() - start_time
    
    print(f"Python列表推导时间: {python_time:.4f}秒")
    print(f"NumPy向量化时间: {numpy_time:.4f}秒")
    print(f"NumPy加速比: {python_time / numpy_time:.2f}倍")

# 4. 使用Numba JIT编译
def numba_optimization():
    # 普通Python函数
    def python_sum(arr):
        total = 0
        for x in arr:
            total += x
        return total
    
    # Numba优化函数
    @jit(nopython=True)
    def numba_sum(arr):
        total = 0
        for x in arr:
            total += x
        return total
    
    # 测试数据
    data = np.random.rand(10000000)
    
    # 预热（第一次运行包含编译时间）
    _ = numba_sum(data)
    
    # 性能测试
    start_time = time.time()
    result1 = python_sum(data)
    python_time = time.time() - start_time
    
    start_time = time.time()
    result2 = numba_sum(data)
    numba_time = time.time() - start_time
    
    print(f"Python循环求和: {python_time:.4f}秒")
    print(f"Numba JIT求和: {numba_time:.4f}秒")
    print(f"Numba加速比: {python_time / numba_time:.2f}倍")

# 5. 内存分析
def memory_analysis():
    tracemalloc.start()
    
    # 内存密集型操作
    def memory_intensive():
        data = []
        for i in range(100000):
            data.append([0] * 1000)  # 创建大量列表
        
        # 模拟一些处理
        result = []
        for sublist in data:
            result.append(sum(sublist))
        
        return result
    
    # 内存优化版本
    def memory_efficient():
        # 使用生成器而不是列表
        def data_generator():
            for i in range(100000):
                yield [0] * 1000
        
        # 流式处理
        total = 0
        for sublist in data_generator():
            total += sum(sublist)
        
        return total
    
    # 分析内存使用
    snapshot1 = tracemalloc.take_snapshot()
    result1 = memory_intensive()
    snapshot2 = tracemalloc.take_snapshot()
    
    # 计算内存差异
    stats = snapshot2.compare_to(snapshot1, 'lineno')
    print("内存密集型函数的内存使用:")
    for stat in stats[:5]:  # 显示前5个最大的内存分配
        print(f"{stat.traceback.format()[-1]}: {stat.size / 1024:.2f} KB")
    
    # 清理内存
    del result1
    gc.collect()
    
    print(f"\n当前内存使用: {tracemalloc.get_traced_memory()[0] / 1024:.2f} KB")
    
    tracemalloc.stop()

# 6. 性能剖析
def profiling_demo():
    def complex_function():
        total = 0
        for i in range(1000):
            for j in range(1000):
                total += i * j
        return total
    
    def another_function():
        data = []
        for i in range(10000):
            data.append(i ** 2)
        return sum(data)
    
    # 使用cProfile进行性能剖析
    profiler = cProfile.Profile()
    profiler.enable()
    
    result1 = complex_function()
    result2 = another_function()
    
    profiler.disable()
    
    # 输出剖析结果
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    print("性能剖析结果:")
    stats.print_stats(10)  # 显示前10个最耗时的函数

# 7. 代码优化技巧
def optimization_tips():
    # 技巧1：使用局部变量
    def use_local_variables():
        import math
        
        # 将全局访问转换为局部变量
        sqrt = math.sqrt
        sin = math.sin
        cos = math.cos
        
        results = []
        for i in range(100000):
            results.append(sqrt(sin(i) ** 2 + cos(i) ** 2))
        
        return results
    
    # 技巧2：避免不必要的属性访问
    class Point:
        def __init__(self, x, y):
            self.x = x
            self.y = y
        
        def distance_slow(self, other):
            # 多次属性访问
            return ((self.x - other.x) ** 2 + 
                   (self.y - other.y) ** 2) ** 0.5
        
        def distance_fast(self, other):
            # 一次性获取属性
            x1, y1 = self.x, self.y
            x2, y2 = other.x, other.y
            return ((x1 - x2) ** 2 + (y1 - y2) ** 2) ** 0.5
    
    # 技巧3：使用生成器表达式
    def generator_vs_list():
        size = 1000000
        
        # 列表推导（消耗内存）
        start_time = time.time()
        sum([x**2 for x in range(size)])
        list_time = time.time() - start_time
        
        # 生成器表达式（节省内存）
        start_time = time.time()
        sum(x**2 for x in range(size))
        gen_time = time.time() - start_time
        
        print(f"列表推导时间: {list_time:.4f}秒")
        print(f"生成器表达式时间: {gen_time:.4f}秒")
        print(f"内存节省比例: 显著")
    
    print("优化技巧演示:")
    use_local_variables()
    
    p1 = Point(1, 2)
    p2 = Point(4, 6)
    print(f"点距离（慢）: {p1.distance_slow(p2)}")
    print(f"点距离（快）: {p1.distance_fast(p2)}")
    
    generator_vs_list()

if __name__ == "__main__":
    print("=== Python性能优化与调试 ===\n")
    
    print("1. 算法优化:")
    algorithm_optimization()
    
    print("\n2. 缓存优化:")
    cache_optimization()
    
    print("\n3. NumPy向量化:")
    numpy_vectorization()
    
    print("\n4. Numba JIT编译:")
    numba_optimization()
    
    print("\n5. 内存分析:")
    memory_analysis()
    
    print("\n6. 性能剖析:")
    profiling_demo()
    
    print("\n7. 代码优化技巧:")
    optimization_tips()

4.3 设计模式与架构

python

from abc import ABC, abstractmethod
from typing import List, Dict, Any
import json
from dataclasses import dataclass
from enum import Enum
import asyncio
from functools import wraps
import time

# 1. 单例模式
class SingletonMeta(type):
    _instances = {}
    
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            instance = super().__call__(*args, **kwargs)
            cls._instances[cls] = instance
        return cls._instances[cls]

class DatabaseConnection(metaclass=SingletonMeta):
    def __init__(self):
        print("创建数据库连接")
        self.connection = "active"
    
    def query(self, sql):
        return f"执行查询: {sql}"

# 2. 工厂模式
class Notification(ABC):
    @abstractmethod
    def send(self, message: str) -> str:
        pass

class EmailNotification(Notification):
    def send(self, message: str) -> str:
        return f"发送邮件: {message}"

class SMSNotification(Notification):
    def send(self, message: str) -> str:
        return f"发送短信: {message}"

class PushNotification(Notification):
    def send(self, message: str) -> str:
        return f"发送推送: {message}"

class NotificationFactory:
    @staticmethod
    def create_notification(notification_type: str) -> Notification:
        if notification_type == "email":
            return EmailNotification()
        elif notification_type == "sms":
            return SMSNotification()
        elif notification_type == "push":
            return PushNotification()
        else:
            raise ValueError(f"未知的通知类型: {notification_type}")

# 3. 观察者模式
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, data):
        for observer in self._observers:
            observer.update(data)

class Observer(ABC):
    @abstractmethod
    def update(self, data):
        pass

class EmailObserver(Observer):
    def update(self, data):
        print(f"邮件观察者收到更新: {data}")

class LogObserver(Observer):
    def update(self, data):
        print(f"日志观察者记录: {data}")

# 4. 策略模式
class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str:
        pass

class CreditCardPayment(PaymentStrategy):
    def pay(self, amount: float) -> str:
        return f"信用卡支付 ${amount:.2f}"

class PayPalPayment(PaymentStrategy):
    def pay(self, amount: float) -> str:
        return f"PayPal支付 ${amount:.2f}"

class BitcoinPayment(PaymentStrategy):
    def pay(self, amount: float) -> str:
        return f"比特币支付 ${amount:.2f}"

class PaymentContext:
    def __init__(self, strategy: PaymentStrategy):
        self._strategy = strategy
    
    def set_strategy(self, strategy: PaymentStrategy):
        self._strategy = strategy
    
    def execute_payment(self, amount: float) -> str:
        return self._strategy.pay(amount)

# 5. 装饰器模式
class DataSource(ABC):
    @abstractmethod
    def write(self, data: str):
        pass
    
    @abstractmethod
    def read(self) -> str:
        pass

class FileDataSource(DataSource):
    def __init__(self, filename: str):
        self.filename = filename
    
    def write(self, data: str):
        with open(self.filename, 'w') as f:
            f.write(data)
        print(f"写入文件: {self.filename}")
    
    def read(self) -> str:
        with open(self.filename, 'r') as f:
            data = f.read()
        print(f"读取文件: {self.filename}")
        return data

class DataSourceDecorator(DataSource):
    def __init__(self, source: DataSource):
        self._wrappee = source
    
    def write(self, data: str):
        self._wrappee.write(data)
    
    def read(self) -> str:
        return self._wrappee.read()

class EncryptionDecorator(DataSourceDecorator):
    def write(self, data: str):
        encrypted = self._encrypt(data)
        super().write(encrypted)
    
    def read(self) -> str:
        encrypted = super().read()
        return self._decrypt(encrypted)
    
    def _encrypt(self, data: str) -> str:
        # 简单的加密示例
        return ''.join(chr(ord(c) + 1) for c in data)
    
    def _decrypt(self, data: str) -> str:
        # 简单的解密示例
        return ''.join(chr(ord(c) - 1) for c in data)

class CompressionDecorator(DataSourceDecorator):
    def write(self, data: str):
        compressed = self._compress(data)
        super().write(compressed)
    
    def read(self) -> str:
        compressed = super().read()
        return self._decompress(compressed)
    
    def _compress(self, data: str) -> str:
        # 简单的压缩示例
        return data.replace('  ', ' ')
    
    def _decompress(self, data: str) -> str:
        # 简单的解压示例
        return data

# 6. 依赖注入
class UserRepository(ABC):
    @abstractmethod
    def get_user(self, user_id: int) -> Dict[str, Any]:
        pass

class MySQLUserRepository(UserRepository):
    def get_user(self, user_id: int) -> Dict[str, Any]:
        # 模拟数据库查询
        return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

class UserService:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def get_user_info(self, user_id: int) -> str:
        user = self.user_repository.get_user(user_id)
        return f"用户: {user['name']}, 邮箱: {user['email']}"

# 7. 命令模式
class Command(ABC):
    @abstractmethod
    def execute(self):
        pass
    
    @abstractmethod
    def undo(self):
        pass

class Light:
    def turn_on(self):
        print("灯已打开")
        return True
    
    def turn_off(self):
        print("灯已关闭")
        return False

class TurnOnLightCommand(Command):
    def __init__(self, light: Light):
        self.light = light
        self.previous_state = None
    
    def execute(self):
        self.previous_state = self.light.turn_off()
        return self.light.turn_on()
    
    def undo(self):
        if self.previous_state:
            return self.light.turn_off()
        return self.light.turn_on()

class RemoteControl:
    def __init__(self):
        self.commands = []
        self.history = []
    
    def set_command(self, command: Command):
        self.commands.append(command)
    
    def press_button(self, index: int):
        if index < len(self.commands):
            result = self.commands[index].execute()
            self.history.append(self.commands[index])
            return result
    
    def undo_last(self):
        if self.history:
            last_command = self.history.pop()
            return last_command.undo()

# 8. 状态模式
class State(ABC):
    @abstractmethod
    def handle(self, context):
        pass

class TrafficLight:
    def __init__(self):
        self.state = RedState()
    
    def change(self):
        self.state.handle(self)
    
    def set_state(self, state: State):
        self.state = state

class RedState(State):
    def handle(self, context: TrafficLight):
        print("红灯 - 停止")
        context.set_state(GreenState())

class GreenState(State):
    def handle(self, context: TrafficLight):
        print("绿灯 - 通行")
        context.set_state(YellowState())

class YellowState(State):
    def handle(self, context: TrafficLight):
        print("黄灯 - 准备")
        context.set_state(RedState())

# 使用示例
def design_patterns_demo():
    print("=== 设计模式演示 ===\n")
    
    # 单例模式
    print("1. 单例模式:")
    db1 = DatabaseConnection()
    db2 = DatabaseConnection()
    print(f"db1 is db2: {db1 is db2}")
    print(f"查询结果: {db1.query('SELECT * FROM users')}")
    
    # 工厂模式
    print("\n2. 工厂模式:")
    factory = NotificationFactory()
    email_notification = factory.create_notification("email")
    sms_notification = factory.create_notification("sms")
    print(email_notification.send("Hello!"))
    print(sms_notification.send("Hi there!"))
    
    # 观察者模式
    print("\n3. 观察者模式:")
    subject = Subject()
    email_observer = EmailObserver()
    log_observer = LogObserver()
    
    subject.attach(email_observer)
    subject.attach(log_observer)
    subject.notify("系统更新完成")
    
    subject.detach(log_observer)
    subject.notify("仅邮件通知")
    
    # 策略模式
    print("\n4. 策略模式:")
    context = PaymentContext(CreditCardPayment())
    print(context.execute_payment(100.50))
    
    context.set_strategy(PayPalPayment())
    print(context.execute_payment(75.25))
    
    # 装饰器模式
    print("\n5. 装饰器模式:")
    source = FileDataSource("test.txt")
    encrypted_source = EncryptionDecorator(source)
    compressed_encrypted_source = CompressionDecorator(encrypted_source)
    
    compressed_encrypted_source.write("Hello World!")
    print(f"读取数据: {compressed_encrypted_source.read()}")
    
    # 依赖注入
    print("\n6. 依赖注入:")
    repository = MySQLUserRepository()
    user_service = UserService(repository)
    print(user_service.get_user_info(1))
    
    # 命令模式
    print("\n7. 命令模式:")
    light = Light()
    turn_on_command = TurnOnLightCommand(light)
    
    remote = RemoteControl()
    remote.set_command(turn_on_command)
    remote.press_button(0)
    remote.undo_last()
    
    # 状态模式
    print("\n8. 状态模式:")
    traffic_light = TrafficLight()
    for _ in range(6):
        traffic_light.change()

if __name__ == "__main__":
    design_patterns_demo()

第五部分：Python学习路径与资源

5.1 学习路径建议

初学者阶段（1-2个月）：

1. 基础语法掌握：变量、数据类型、控制流、函数

2. 核心数据结构：列表、字典、集合、元组

3. 文件操作：读写文本文件、CSV、JSON

4. 面向对象编程：类、对象、继承、多态

5. 错误处理：异常处理、调试技巧

中级阶段（3-6个月）：

1. 函数式编程：高阶函数、装饰器、生成器

2. 模块与包：创建和使用模块、虚拟环境

3. 常用标准库：os、sys、datetime、collections、itertools

4. 数据库操作：SQLite、MySQL连接

5. Web基础：Flask或Django入门

高级阶段（6-12个月）：

1. 并发编程：多线程、多进程、异步IO

2. 网络编程：socket、HTTP协议、RESTful API

3. 性能优化：算法优化、内存管理、性能剖析

4. 设计模式：常用设计模式实现

5. 元编程：装饰器、元类、描述符

专业方向（1年以上）：

1. 数据科学：NumPy、Pandas、Matplotlib、Scikit-learn

2. Web开发：Django、FastAPI、数据库设计、前端集成

3. 人工智能：TensorFlow、PyTorch、自然语言处理

4. 自动化运维：Ansible、Docker、Kubernetes

5. 区块链开发：Web3.py、智能合约

5.2 优质学习资源

官方文档：

• Python官方文档

• PEP索引

在线课程：

• Coursera：密歇根大学Python专项课程

• edX：MIT计算机科学导论（Python版）

• Udemy：Complete Python Bootcamp

• 中国大学MOOC：北京大学Python课程

书籍推荐：

• 入门：《Python编程：从入门到实践》

• 进阶：《流畅的Python》

• 算法：《Python算法教程》

• 数据分析：《利用Python进行数据分析》

• Web开发：《Django企业开发实战》

实战项目：

1. 命令行工具：待办事项应用、文件管理器

2. Web应用：博客系统、电子商务平台

3. 数据分析：股票分析、用户行为分析

4. 机器学习：图像分类、推荐系统

5. 自动化脚本：文件批量处理、网站爬虫

5.3 社区与求职

技术社区：

• Stack Overflow：问题解答

• GitHub：开源项目贡献

• Reddit：r/Python、r/learnpython

• 中文社区：Python中文社区、V2EX

求职建议：

1. 构建作品集：GitHub项目、技术博客

2. 准备面试：算法题、系统设计、项目经验

3. 关注趋势：新框架、新技术、行业动态

4. 持续学习：参加技术会议、阅读源码

5. 建立网络：LinkedIn、技术社区、线下活动

5.4 Python未来发展趋势

热门方向：

1. 人工智能与机器学习：模型部署、边缘计算

2. 数据科学与大数据：实时分析、数据管道

3. Web3与区块链：智能合约、DApp开发

4. 自动化与DevOps：基础设施即代码、CI/CD

5. 量化金融：算法交易、风险分析

技术趋势：

1. 类型注解普及：更好的代码可读性和工具支持

2. 异步编程主流化：高性能Web服务和实时应用

3. 容器化部署：Docker、Kubernetes成为标准

4. 无服务器架构：函数即服务（FaaS）

5. 跨平台开发：移动应用、桌面应用

结语

Python作为一门多功能、易学且强大的编程语言，在各个领域都展现出卓越的表现。从基础的脚本编写到复杂的人工智能系统，Python都能提供合适的工具和库支持。

学习Python不仅仅是学习一门语言，更是学习一种解决问题的思维方式。通过掌握Python，你将能够：

1. 自动化日常工作，提高效率

2. 分析数据，发现有价值的洞见

3. 构建Web应用，实现创意想法

4. 开发智能系统，解决复杂问题

5. 加入全球开发者社区，共同推动技术进步

无论你是编程新手还是经验丰富的开发者，Python都能为你打开新的大门。最重要的是保持学习的热情和持续的实践。编程是一门实践性很强的技能，只有通过不断的编码、调试和优化，才能真正掌握Python的精髓。

记住，学习编程的旅程永远不会真正结束，因为技术总是在不断演进。但只要你掌握了核心概念和持续学习的能力，就能在技术世界中游刃有余。

开始你的Python之旅吧，编码快乐！ 🐍✨

---

本指南涵盖了Python从入门到精通的各个方面，总计超过15000字。内容从基础语法到高级主题，再到实际应用领域，提供了全面的学习路径和实践示例。建议读者按照自己的节奏学习，多做实践项目，逐步深入理解Python的各个层面。