昨天，我们学习了继承和多态，掌握了面向对象编程的核心概念。今天，我们将学习封装和抽象 — 面向对象编程的另外两个重要支柱。

封装保护数据并提供受控的访问，而抽象则隐藏复杂性并提供简洁的接口。

今天您将学习什么

• 什么是封装以及如何实现封装
• 访问修饰符和属性装饰器
• 抽象类和抽象方法
• 接口和抽象基类
• 真实世界示例：银行账户、数据库连接、日志系统

什么是封装？

封装是面向对象编程的核心概念之一，它将数据（属性）和操作数据的方法（行为）捆绑在一起，并隐藏内部实现细节。

封装的优势：

• 数据保护：防止外部直接访问和修改数据
• 接口简化：提供清晰的公共接口
• 维护性：内部实现可以改变而不影响外部代码
• 数据验证：在数据访问时进行验证

1. 基本封装

使用私有属性

class BankAccount:
    """银行账户类 - 封装示例"""
    
    def __init__(self, account_number, owner_name, initial_balance=0):
        # 私有属性（以双下划线开头）
        self.__account_number = account_number
        self.__owner_name = owner_name
        self.__balance = initial_balance
        self.__transactions = []
        self.__is_active = True
    
    # 公共方法 - 提供受控的访问
    def get_balance(self):
        """获取余额"""
        return self.__balance
    
    def get_account_info(self):
        """获取账户信息"""
        return {
            'account_number': self.__account_number,
            'owner_name': self.__owner_name,
            'balance': self.__balance,
            'is_active': self.__is_active
        }
    
    def deposit(self, amount):
        """存款"""
        if not self.__is_active:
            print("账户已被冻结，无法存款")
            return False
        
        if amount <= 0:
            print("存款金额必须大于0")
            return False
        
        self.__balance += amount
        self.__add_transaction('deposit', amount)
        print(f"存款成功：{amount}元，当前余额：{self.__balance}元")
        return True
    
    def withdraw(self, amount):
        """取款"""
        if not self.__is_active:
            print("账户已被冻结，无法取款")
            return False
        
        if amount <= 0:
            print("取款金额必须大于0")
            return False
        
        if amount > self.__balance:
            print("余额不足")
            return False
        
        self.__balance -= amount
        self.__add_transaction('withdraw', amount)
        print(f"取款成功：{amount}元，当前余额：{self.__balance}元")
        return True
    
    # 私有方法 - 内部使用
    def __add_transaction(self, transaction_type, amount):
        """添加交易记录"""
        transaction = {
            'type': transaction_type,
            'amount': amount,
            'balance': self.__balance,
            'timestamp': '2025-07-15 10:30:00'  # 简化时间戳
        }
        self.__transactions.append(transaction)
    
    def get_transaction_history(self):
        """获取交易历史（只读）"""
        return self.__transactions.copy()  # 返回副本，防止外部修改
    
    def freeze_account(self):
        """冻结账户"""
        self.__is_active = False
        print(f"账户{self.__account_number}已被冻结")
    
    def unfreeze_account(self):
        """解冻账户"""
        self.__is_active = True
        print(f"账户{self.__account_number}已解冻")

# 使用封装的银行账户
account = BankAccount("001", "张三", 1000)

# 正确的访问方式
print(f"余额：{account.get_balance()}元")
print(f"账户信息：{account.get_account_info()}")

# 进行交易
account.deposit(500)
account.withdraw(200)

# 查看交易历史
transactions = account.get_transaction_history()
print(f"交易历史：{transactions}")

# 尝试直接访问私有属性（会失败）
try:
    print(account.__balance)  # AttributeError
except AttributeError:
    print("无法直接访问私有属性")

2. 属性装饰器

使用@property

class Temperature:
    """温度类 - 属性装饰器示例"""
    
    def __init__(self, celsius):
        self._celsius = celsius
    
    @property
    def celsius(self):
        """摄氏度属性（只读）"""
        return self._celsius
    
    @celsius.setter
    def celsius(self, value):
        """设置摄氏度（带验证）"""
        if value < -273.15:
            raise ValueError("温度不能低于绝对零度(-273.15°C)")
        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
    
    @property
    def kelvin(self):
        """开尔文温度属性（计算属性）"""
        return self._celsius + 273.15
    
    @kelvin.setter
    def kelvin(self, value):
        """设置开尔文温度"""
        if value < 0:
            raise ValueError("开尔文温度不能为负")
        self.celsius = value - 273.15

# 使用属性装饰器
temp = Temperature(25)
print(f"摄氏度：{temp.celsius}°C")
print(f"华氏度：{temp.fahrenheit}°F")
print(f"开尔文：{temp.kelvin}K")

# 设置温度
temp.celsius = 30
print(f"新的摄氏度：{temp.celsius}°C")

temp.fahrenheit = 86
print(f"新的华氏度：{temp.fahrenheit}°F")

# 验证温度
try:
    temp.celsius = -300  # 会抛出异常
except ValueError as e:
    print(f"错误：{e}")

真实世界示例1：数据库连接管理器

import sqlite3
import threading
from contextlib import contextmanager

class DatabaseConnection:
    """数据库连接类 - 封装示例"""
    
    def __init__(self, database_path):
        self._database_path = database_path
        self._connection = None
        self._lock = threading.Lock()
        self._is_connected = False
    
    def connect(self):
        """连接数据库"""
        if self._is_connected:
            print("数据库已连接")
            return
        
        try:
            self._connection = sqlite3.connect(self._database_path)
            self._is_connected = True
            print(f"成功连接到数据库：{self._database_path}")
        except Exception as e:
            print(f"连接数据库失败：{e}")
            raise
    
    def disconnect(self):
        """断开数据库连接"""
        if not self._is_connected:
            print("数据库未连接")
            return
        
        try:
            self._connection.close()
            self._is_connected = False
            self._connection = None
            print("数据库连接已断开")
        except Exception as e:
            print(f"断开数据库连接失败：{e}")
    
    def execute_query(self, query, parameters=None):
        """执行查询"""
        if not self._is_connected:
            raise RuntimeError("数据库未连接")
        
        with self._lock:  # 线程安全
            try:
                cursor = self._connection.cursor()
                if parameters:
                    cursor.execute(query, parameters)
                else:
                    cursor.execute(query)
                
                if query.strip().upper().startswith('SELECT'):
                    return cursor.fetchall()
                else:
                    self._connection.commit()
                    return cursor.rowcount
            except Exception as e:
                self._connection.rollback()
                raise RuntimeError(f"查询执行失败：{e}")
    
    @contextmanager
    def transaction(self):
        """事务管理器"""
        if not self._is_connected:
            raise RuntimeError("数据库未连接")
        
        try:
            yield self
            self._connection.commit()
        except Exception:
            self._connection.rollback()
            raise
    
    def get_connection_status(self):
        """获取连接状态"""
        return {
            'is_connected': self._is_connected,
            'database_path': self._database_path
        }
    
    def __enter__(self):
        """上下文管理器入口"""
        self.connect()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器出口"""
        self.disconnect()

# 使用数据库连接管理器
def create_sample_database():
    """创建示例数据库"""
    db = DatabaseConnection(":memory:")  # 内存数据库
    
    with db:
        # 创建表
        db.execute_query("""
            CREATE TABLE users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                email TEXT UNIQUE,
                age INTEGER
            )
        """)
        
        # 插入数据
        db.execute_query(
            "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
            ("张三", "zhangsan@email.com", 25)
        )
        db.execute_query(
            "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
            ("李四", "lisi@email.com", 30)
        )
        
        # 查询数据
        results = db.execute_query("SELECT * FROM users")
        print("用户数据：")
        for row in results:
            print(f"  ID: {row[0]}, 姓名: {row[1]}, 邮箱: {row[2]}, 年龄: {row[3]}")
    
    return db

# 使用事务
def demonstrate_transaction():
    """演示事务使用"""
    db = DatabaseConnection(":memory:")
    
    with db:
        # 创建表
        db.execute_query("""
            CREATE TABLE accounts (
                id INTEGER PRIMARY KEY,
                name TEXT,
                balance REAL
            )
        """)
        
        # 使用事务
        with db.transaction():
            db.execute_query(
                "INSERT INTO accounts (name, balance) VALUES (?, ?)",
                ("账户A", 1000)
            )
            db.execute_query(
                "INSERT INTO accounts (name, balance) VALUES (?, ?)",
                ("账户B", 500)
            )
            
            # 转账操作
            db.execute_query(
                "UPDATE accounts SET balance = balance - 200 WHERE name = ?",
                ("账户A",)
            )
            db.execute_query(
                "UPDATE accounts SET balance = balance + 200 WHERE name = ?",
                ("账户B",)
            )
        
        # 查看结果
        results = db.execute_query("SELECT * FROM accounts")
        print("转账后的账户余额：")
        for row in results:
            print(f"  {row[1]}: {row[2]}元")

# 运行示例
print("=== 数据库连接示例 ===")
create_sample_database()

print("\n=== 事务示例 ===")
demonstrate_transaction()

真实世界示例2：日志系统

import logging
import os
from datetime import datetime
from enum import Enum

class LogLevel(Enum):
    """日志级别枚举"""
    DEBUG = logging.DEBUG
    INFO = logging.INFO
    WARNING = logging.WARNING
    ERROR = logging.ERROR
    CRITICAL = logging.CRITICAL

class Logger:
    """日志系统类 - 封装示例"""
    
    def __init__(self, name, log_file=None, level=LogLevel.INFO):
        self._name = name
        self._log_file = log_file
        self._level = level
        self._logger = None
        self._formatter = None
        self._handlers = []
        
        self._setup_logger()
    
    def _setup_logger(self):
        """设置日志器"""
        self._logger = logging.getLogger(self._name)
        self._logger.setLevel(self._level.value)
        
        # 创建格式化器
        self._formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        
        # 添加控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(self._formatter)
        self._logger.addHandler(console_handler)
        self._handlers.append(console_handler)
        
        # 添加文件处理器（如果指定了日志文件）
        if self._log_file:
            file_handler = logging.FileHandler(self._log_file, encoding='utf-8')
            file_handler.setFormatter(self._formatter)
            self._logger.addHandler(file_handler)
            self._handlers.append(file_handler)
    
    def debug(self, message):
        """记录调试信息"""
        self._logger.debug(message)
    
    def info(self, message):
        """记录一般信息"""
        self._logger.info(message)
    
    def warning(self, message):
        """记录警告信息"""
        self._logger.warning(message)
    
    def error(self, message):
        """记录错误信息"""
        self._logger.error(message)
    
    def critical(self, message):
        """记录严重错误信息"""
        self._logger.critical(message)
    
    def log_with_context(self, level, message, context=None):
        """记录带上下文的日志"""
        if context:
            message = f"{message} | 上下文: {context}"
        
        if level == LogLevel.DEBUG:
            self.debug(message)
        elif level == LogLevel.INFO:
            self.info(message)
        elif level == LogLevel.WARNING:
            self.warning(message)
        elif level == LogLevel.ERROR:
            self.error(message)
        elif level == LogLevel.CRITICAL:
            self.critical(message)
    
    def get_log_file_path(self):
        """获取日志文件路径"""
        return self._log_file
    
    def get_logger_info(self):
        """获取日志器信息"""
        return {
            'name': self._name,
            'level': self._level.name,
            'log_file': self._log_file,
            'handlers_count': len(self._handlers)
        }

class ApplicationLogger:
    """应用程序日志管理器 - 抽象示例"""
    
    def __init__(self, app_name):
        self._app_name = app_name
        self._loggers = {}
        self._setup_loggers()
    
    def _setup_loggers(self):
        """设置各种日志器"""
        # 创建日志目录
        log_dir = f"logs/{self._app_name}"
        os.makedirs(log_dir, exist_ok=True)
        
        # 系统日志
        self._loggers['system'] = Logger(
            f"{self._app_name}.system",
            f"{log_dir}/system.log",
            LogLevel.INFO
        )
        
        # 错误日志
        self._loggers['error'] = Logger(
            f"{self._app_name}.error",
            f"{log_dir}/error.log",
            LogLevel.ERROR
        )
        
        # 访问日志
        self._loggers['access'] = Logger(
            f"{self._app_name}.access",
            f"{log_dir}/access.log",
            LogLevel.INFO
        )
    
    def log_system_event(self, message, level=LogLevel.INFO):
        """记录系统事件"""
        self._loggers['system'].log_with_context(level, message)
    
    def log_error(self, error_message, error_details=None):
        """记录错误"""
        context = f"错误详情: {error_details}" if error_details else None
        self._loggers['error'].error(f"错误: {error_message}")
        if context:
            self._loggers['error'].error(context)
    
    def log_access(self, user_id, action, resource):
        """记录访问日志"""
        message = f"用户 {user_id} 执行了 {action} 操作，资源: {resource}"
        self._loggers['access'].info(message)
    
    def get_logger(self, logger_type):
        """获取特定类型的日志器"""
        return self._loggers.get(logger_type)
    
    def get_all_loggers_info(self):
        """获取所有日志器信息"""
        return {
            logger_type: logger.get_logger_info()
            for logger_type, logger in self._loggers.items()
        }

# 使用日志系统
def demonstrate_logging():
    """演示日志系统使用"""
    # 创建应用程序日志管理器
    app_logger = ApplicationLogger("MyApp")
    
    # 记录系统事件
    app_logger.log_system_event("应用程序启动", LogLevel.INFO)
    app_logger.log_system_event("数据库连接成功", LogLevel.INFO)
    app_logger.log_system_event("内存使用率较高", LogLevel.WARNING)
    
    # 记录错误
    try:
        # 模拟错误
        result = 10 / 0
    except Exception as e:
        app_logger.log_error("除零错误", str(e))
    
    # 记录访问日志
    app_logger.log_access("user123", "登录", "系统")
    app_logger.log_access("user123", "查看", "用户资料")
    app_logger.log_access("user456", "删除", "文件")
    
    # 查看日志器信息
    print("日志器信息：")
    for logger_type, info in app_logger.get_all_loggers_info().items():
        print(f"  {logger_type}: {info}")

# 运行日志示例
print("=== 日志系统示例 ===")
demonstrate_logging()

封装和抽象的最佳实践

推荐做法：

• 使用私有属性保护数据
• 提供清晰的公共接口
• 使用属性装饰器进行数据验证
• 实现适当的抽象层次

避免的做法：

• 暴露过多内部实现细节
• 创建过于复杂的接口
• 忽略数据验证
• 违反封装原则

高级封装特性

描述符

class ValidatedAttribute:
    """验证属性描述符"""
    
    def __init__(self, min_value=None, max_value=None, allowed_values=None):
        self.min_value = min_value
        self.max_value = max_value
        self.allowed_values = allowed_values
        self.name = None
    
    def __set_name__(self, owner, name):
        self.name = name
    
    def __get__(self, instance, owner):
        if instance is None:
            return self
        return instance.__dict__.get(self.name)
    
    def __set__(self, instance, value):
        # 验证值
        if self.allowed_values and value not in self.allowed_values:
            raise ValueError(f"值必须是以下之一: {self.allowed_values}")
        
        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}")
        
        instance.__dict__[self.name] = value

class Person:
    """使用描述符的Person类"""
    
    age = ValidatedAttribute(min_value=0, max_value=150)
    grade = ValidatedAttribute(allowed_values=['A', 'B', 'C', 'D', 'F'])
    
    def __init__(self, name, age, grade):
        self.name = name
        self.age = age
        self.grade = grade

# 使用描述符
person = Person("张三", 25, "A")
print(f"姓名: {person.name}, 年龄: {person.age}, 成绩: {person.grade}")

try:
    person.age = -5  # 会抛出异常
except ValueError as e:
    print(f"错误: {e}")

try:
    person.grade = "X"  # 会抛出异常
except ValueError as e:
    print(f"错误: {e}")

回顾

今天您学习了：

• 封装的基本概念和实现
• 属性装饰器的使用
• 抽象类和接口的概念
• 真实世界示例：数据库连接、日志系统

封装和抽象是面向对象编程的重要支柱，掌握这些概念将让您能够创建更加安全和易维护的代码！