跳到主要内容

Python进阶笔记

1. 面向对象编程​

类的定义和实例化​

class Person:
    # 类变量
    species = "Homo sapiens"
  
    def __init__(self, name, age):
        # 实例变量
        self.name = name
        self.age = age
  
    # 实例方法
    def introduce(self):
        return f"我叫{self.name}，今年{self.age}岁"
  
    # 类方法
    @classmethod
    def from_birth_year(cls, name, birth_year):
        age = 2024 - birth_year
        return cls(name, age)
  
    # 静态方法
    @staticmethod
    def is_adult(age):
        return age >= 18

# 使用类
person1 = Person("张三", 25)
person2 = Person.from_birth_year("李四", 1995)

print(person1.introduce())
print(Person.is_adult(16))  # False

继承​

# 父类
class Animal:
    def __init__(self, name, species):
        self.name = name
        self.species = species
  
    def speak(self):
        pass
  
    def info(self):
        return f"{self.name}是一只{self.species}"

# 子类
class Dog(Animal):
    def __init__(self, name, breed):
        super().__init__(name, "狗")
        self.breed = breed
  
    def speak(self):
        return "汪汪！"
  
    def fetch(self):
        return f"{self.name}正在捡球"

class Cat(Animal):
    def __init__(self, name):
        super().__init__(name, "猫")
  
    def speak(self):
        return "喵喵！"
  
    def climb(self):
        return f"{self.name}正在爬树"

# 使用继承
dog = Dog("旺财", "金毛")
cat = Cat("咪咪")

print(dog.info())      # 旺财是一只狗
print(dog.speak())     # 汪汪！
print(cat.speak())     # 喵喵！

多重继承​

class Flyable:
    def fly(self):
        return "正在飞行"

class Swimmable:
    def swim(self):
        return "正在游泳"

class Duck(Animal, Flyable, Swimmable):
    def __init__(self, name):
        super().__init__(name, "鸭子")
  
    def speak(self):
        return "嘎嘎！"

# 使用多重继承
duck = Duck("唐老鸭")
print(duck.speak())    # 嘎嘎！
print(duck.fly())      # 正在飞行
print(duck.swim())     # 正在游泳

封装和属性​

class BankAccount:
    def __init__(self, account_number, initial_balance=0):
        self.account_number = account_number
        self._balance = initial_balance  # 受保护的属性
        self.__pin = "1234"             # 私有属性
  
    @property
    def balance(self):
        """余额属性的getter"""
        return self._balance
  
    @balance.setter
    def balance(self, value):
        """余额属性的setter"""
        if value < 0:
            raise ValueError("余额不能为负数")
        self._balance = value
  
    def deposit(self, amount):
        """存款"""
        if amount > 0:
            self._balance += amount
            return f"存款成功，当前余额：{self._balance}"
        return "存款金额必须大于0"
  
    def withdraw(self, amount, pin):
        """取款"""
        if pin != self.__pin:
            return "密码错误"
        if amount > self._balance:
            return "余额不足"
        self._balance -= amount
        return f"取款成功，当前余额：{self._balance}"

# 使用封装
account = BankAccount("123456789", 1000)
print(account.balance)          # 1000
print(account.deposit(500))     # 存款成功，当前余额：1500
print(account.withdraw(200, "1234"))  # 取款成功，当前余额：1300

特殊方法（魔法方法）​

class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y
  
    def __str__(self):
        """字符串表示"""
        return f"Vector({self.x}, {self.y})"
  
    def __repr__(self):
        """开发者字符串表示"""
        return f"Vector({self.x}, {self.y})"
  
    def __add__(self, other):
        """加法运算"""
        return Vector(self.x + other.x, self.y + other.y)
  
    def __mul__(self, scalar):
        """标量乘法"""
        return Vector(self.x * scalar, self.y * scalar)
  
    def __eq__(self, other):
        """相等比较"""
        return self.x == other.x and self.y == other.y
  
    def __len__(self):
        """向量长度"""
        return int((self.x ** 2 + self.y ** 2) ** 0.5)

# 使用特殊方法
v1 = Vector(1, 2)
v2 = Vector(3, 4)

print(v1)           # Vector(1, 2)
print(v1 + v2)      # Vector(4, 6)
print(v1 * 2)       # Vector(2, 4)
print(v1 == v2)     # False
print(len(v1))      # 2

2. 高级特性​

装饰器​

函数装饰器​

import time
import functools

# 简单装饰器
def timer(func):
    """计时装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 执行时间: {end - start:.4f}秒")
        return result
    return wrapper

# 带参数的装饰器
def repeat(times):
    """重复执行装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

# 使用装饰器
@timer
def slow_function():
    time.sleep(1)
    return "完成"

@repeat(3)
def greet(name):
    print(f"Hello, {name}!")

# 调用
slow_function()
greet("世界")

类装饰器​

class CountCalls:
    """统计函数调用次数的类装饰器"""
    def __init__(self, func):
        self.func = func
        self.count = 0
  
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"{self.func.__name__} 被调用了 {self.count} 次")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello():
    print("Hello!")

# 调用
say_hello()  # say_hello 被调用了 1 次
say_hello()  # say_hello 被调用了 2 次

生成器和迭代器​

生成器​

# 生成器函数
def fibonacci(n):
    """斐波那契数列生成器"""
    a, b = 0, 1
    count = 0
    while count < n:
        yield a
        a, b = b, a + b
        count += 1

# 生成器表达式
squares = (x**2 for x in range(10))

# 使用生成器
fib_gen = fibonacci(10)
for num in fib_gen:
    print(num, end=" ")  # 0 1 1 2 3 5 8 13 21 34

# 大文件处理示例
def read_large_file(file_path):
    """逐行读取大文件"""
    with open(file_path, 'r', encoding='utf-8') as file:
        for line in file:
            yield line.strip()

自定义迭代器​

class Countdown:
    """倒计时迭代器"""
    def __init__(self, start):
        self.start = start
  
    def __iter__(self):
        return self
  
    def __next__(self):
        if self.start <= 0:
            raise StopIteration
        self.start -= 1
        return self.start + 1

# 使用自定义迭代器
countdown = Countdown(5)
for num in countdown:
    print(num)  # 5 4 3 2 1

上下文管理器​

# 使用contextlib
from contextlib import contextmanager

@contextmanager
def timer_context():
    """计时上下文管理器"""
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f"执行时间: {end - start:.4f}秒")

# 自定义上下文管理器类
class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
  
    def __enter__(self):
        print(f"打开文件: {self.filename}")
        self.file = open(self.filename, self.mode)
        return self.file
  
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"关闭文件: {self.filename}")
        if self.file:
            self.file.close()

# 使用上下文管理器
with timer_context():
    time.sleep(1)

with FileManager("test.txt", "w") as f:
    f.write("Hello, World!")

元编程​

元类​

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

class Database(metaclass=SingletonMeta):
    def __init__(self):
        print("创建数据库连接")

# 测试单例
db1 = Database()  # 创建数据库连接
db2 = Database()  # 不会再次创建
print(db1 is db2)  # True

动态创建类​

# 使用type动态创建类
def init_method(self, name):
    self.name = name

def greet_method(self):
    return f"Hello, I'm {self.name}"

# 动态创建Person类
Person = type('Person', (object,), {
    '__init__': init_method,
    'greet': greet_method
})

person = Person("张三")
print(person.greet())  # Hello, I'm 张三

描述符​

class ValidatedAttribute:
    """验证属性描述符"""
    def __init__(self, name, validator):
        self.name = name
        self.validator = validator
  
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, f'_{self.name}')
  
    def __set__(self, obj, value):
        if not self.validator(value):
            raise ValueError(f"Invalid value for {self.name}: {value}")
        setattr(obj, f'_{self.name}', value)

class Person:
    name = ValidatedAttribute('name', lambda x: isinstance(x, str) and len(x) > 0)
    age = ValidatedAttribute('age', lambda x: isinstance(x, int) and 0 <= x <= 150)
  
    def __init__(self, name, age):
        self.name = name
        self.age = age

# 使用描述符
person = Person("张三", 25)
# person.age = -5  # 会抛出ValueError

3. 函数式编程​

高阶函数​

# map, filter, reduce
from functools import reduce

numbers = [1, 2, 3, 4, 5]

# map - 映射
squares = list(map(lambda x: x**2, numbers))
print(squares)  # [1, 4, 9, 16, 25]

# filter - 过滤
evens = list(filter(lambda x: x % 2 == 0, numbers))
print(evens)  # [2, 4]

# reduce - 归约
sum_all = reduce(lambda x, y: x + y, numbers)
print(sum_all)  # 15

# 自定义高阶函数
def apply_operation(func, *args):
    """应用操作到所有参数"""
    return [func(arg) for arg in args]

def compose(f, g):
    """函数组合"""
    return lambda x: f(g(x))

# 使用
results = apply_operation(lambda x: x**2, 1, 2, 3, 4)
print(results)  # [1, 4, 9, 16]

double_then_square = compose(lambda x: x**2, lambda x: x*2)
print(double_then_square(3))  # 36

偏函数​

from functools import partial

# 创建偏函数
def multiply(x, y, z):
    return x * y * z

# 固定前两个参数
double_triple = partial(multiply, 2, 3)
print(double_triple(4))  # 24

# 日志函数示例
def log(level, message):
    print(f"[{level}] {message}")

# 创建特定级别的日志函数
info = partial(log, "INFO")
error = partial(log, "ERROR")

info("程序启动")    # [INFO] 程序启动
error("发生错误")   # [ERROR] 发生错误

4. 异步编程​

asyncio基础​

import asyncio
import aiohttp

# 基本异步函数
async def say_hello(name, delay):
    print(f"开始问候 {name}")
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")
    return f"问候了 {name}"

# 并发执行
async def main():
    # 并发执行多个异步任务
    tasks = [
        say_hello("张三", 1),
        say_hello("李四", 2),
        say_hello("王五", 1.5)
    ]
  
    results = await asyncio.gather(*tasks)
    print("所有任务完成:", results)

# 运行异步程序
# asyncio.run(main())

异步HTTP请求​

async def fetch_url(session, url):
    """异步获取URL内容"""
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"Error fetching {url}: {e}"

async def fetch_multiple_urls(urls):
    """并发获取多个URL"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# 使用示例
urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/1"
]

# results = asyncio.run(fetch_multiple_urls(urls))

5. 设计模式​

单例模式​

class Singleton:
    _instance = None
    _initialized = False
  
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
  
    def __init__(self):
        if not self._initialized:
            print("初始化单例")
            self._initialized = True

# 装饰器实现单例
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 DatabaseConnection:
    def __init__(self):
        print("创建数据库连接")

工厂模式​

from abc import ABC, abstractmethod

# 抽象产品
class Animal(ABC):
    @abstractmethod
    def speak(self):
        pass

# 具体产品
class Dog(Animal):
    def speak(self):
        return "汪汪！"

class Cat(Animal):
    def speak(self):
        return "喵喵！"

# 工厂
class AnimalFactory:
    @staticmethod
    def create_animal(animal_type):
        if animal_type.lower() == "dog":
            return Dog()
        elif animal_type.lower() == "cat":
            return Cat()
        else:
            raise ValueError(f"未知的动物类型: {animal_type}")

# 使用工厂
factory = AnimalFactory()
dog = factory.create_animal("dog")
cat = factory.create_animal("cat")
print(dog.speak())  # 汪汪！
print(cat.speak())  # 喵喵！

观察者模式​

class Subject:
    """被观察者"""
    def __init__(self):
        self._observers = []
        self._state = None
  
    def attach(self, observer):
        """添加观察者"""
        self._observers.append(observer)
  
    def detach(self, observer):
        """移除观察者"""
        self._observers.remove(observer)
  
    def notify(self):
        """通知所有观察者"""
        for observer in self._observers:
            observer.update(self._state)
  
    def set_state(self, state):
        """设置状态并通知观察者"""
        self._state = state
        self.notify()

class Observer:
    """观察者"""
    def __init__(self, name):
        self.name = name
  
    def update(self, state):
        print(f"{self.name} 收到状态更新: {state}")

# 使用观察者模式
subject = Subject()
observer1 = Observer("观察者1")
observer2 = Observer("观察者2")

subject.attach(observer1)
subject.attach(observer2)

subject.set_state("新状态")  # 两个观察者都会收到通知

6. 性能优化​

使用__slots__​

class RegularClass:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class OptimizedClass:
    __slots__ = ['x', 'y']  # 限制属性，节省内存
  
    def __init__(self, x, y):
        self.x = x
        self.y = y

缓存和记忆化​

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    """使用缓存的斐波那契函数"""
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 自定义缓存装饰器
def memo(func):
    cache = {}
    def wrapper(*args):
        if args in cache:
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return wrapper

@memo
def expensive_function(n):
    """昂贵的计算函数"""
    print(f"计算 {n}")
    return n ** n

性能测量​

import cProfile
import timeit

# 使用timeit测量性能
def test_list_comprehension():
    return [x**2 for x in range(1000)]

def test_map():
    return list(map(lambda x: x**2, range(1000)))

# 比较性能
time1 = timeit.timeit(test_list_comprehension, number=10000)
time2 = timeit.timeit(test_map, number=10000)

print(f"列表推导式: {time1:.4f}秒")
print(f"map函数: {time2:.4f}秒")

# 使用cProfile分析
def profile_example():
    # 一些复杂的操作
    data = [i**2 for i in range(10000)]
    return sum(data)

cProfile.run('profile_example()')

7. 实践项目​

简单的Web爬虫​

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import time

class WebScraper:
    def __init__(self, max_concurrent=10):
        self.max_concurrent = max_concurrent
        self.session = None
  
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
  
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
  
    async def fetch_page(self, url):
        """异步获取网页内容"""
        try:
            async with self.session.get(url) as response:
                return await response.text()
        except Exception as e:
            print(f"获取 {url} 失败: {e}")
            return None
  
    async def parse_content(self, html, url):
        """解析网页内容"""
        if not html:
            return None
      
        soup = BeautifulSoup(html, 'html.parser')
        return {
            'url': url,
            'title': soup.title.string if soup.title else '无标题',
            'links': [a.get('href') for a in soup.find_all('a', href=True)]
        }
  
    async def scrape_urls(self, urls):
        """批量爬取URL"""
        semaphore = asyncio.Semaphore(self.max_concurrent)
      
        async def scrape_single(url):
            async with semaphore:
                html = await self.fetch_page(url)
                return await self.parse_content(html, url)
      
        tasks = [scrape_single(url) for url in urls]
        results = await asyncio.gather(*tasks)
        return [result for result in results if result]

# 使用爬虫
async def main():
    urls = [
        'https://example.com',
        'https://httpbin.org',
        # 更多URL...
    ]
  
    async with WebScraper() as scraper:
        results = await scraper.scrape_urls(urls)
        for result in results:
            print(f"标题: {result['title']}")
            print(f"URL: {result['url']}")
            print(f"链接数量: {len(result['links'])}")
            print("-" * 50)

# asyncio.run(main())

装饰器应用：API限流​

import time
from functools import wraps
from collections import defaultdict, deque

class RateLimiter:
    def __init__(self, max_calls=10, time_window=60):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = defaultdict(deque)
  
    def is_allowed(self, key):
        now = time.time()
        call_times = self.calls[key]
      
        # 清理过期的调用记录
        while call_times and call_times[0] <= now - self.time_window:
            call_times.popleft()
      
        # 检查是否超过限制
        if len(call_times) >= self.max_calls:
            return False
      
        # 记录本次调用
        call_times.append(now)
        return True

def rate_limit(max_calls=10, time_window=60, key_func=None):
    """API限流装饰器"""
    limiter = RateLimiter(max_calls, time_window)
  
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 确定限流的key
            if key_func:
                key = key_func(*args, **kwargs)
            else:
                key = "default"
          
            if not limiter.is_allowed(key):
                raise Exception(f"Rate limit exceeded for {key}")
          
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 使用限流装饰器
@rate_limit(max_calls=5, time_window=10)
def api_call(user_id):
    return f"API调用成功，用户ID: {user_id}"

# 测试
for i in range(7):
    try:
        result = api_call("user123")
        print(result)
    except Exception as e:
        print(f"调用失败: {e}")
    time.sleep(1)

总结​

Python进阶特性包括：

核心概念​

  • 面向对象编程：继承、封装、多态、特殊方法
  • 函数式编程：高阶函数、装饰器、生成器
  • 异步编程：async/await、协程、并发处理

高级特性​

  • 元编程：元类、描述符、动态类创建
  • 上下文管理器：资源管理、异常安全
  • 性能优化：缓存、slots、性能分析

设计模式​

  • 创建型：单例、工厂模式
  • 行为型：观察者、装饰器模式
  • 结构型：适配器、代理模式

实践建议​

  1. 逐步学习：从基础OOP开始，逐步深入
  2. 实践项目：通过实际项目巩固知识
  3. 代码审查：学习优秀开源项目的代码
  4. 性能意识：关注代码性能和最佳实践