Python脚本工厂模式适用哪些场景

wen 实用脚本 2026-07-19 4

本文目录导读：

Python脚本工厂模式适用哪些场景

对象创建逻辑复杂
需要解耦客户端和具体类
需要动态选择算法/策略
数据库/数据源切换
日志/监控系统集成
插件系统
测试模拟
资源池管理
不适合使用的场景：
Python特有的简单替代方案：

工厂模式（Factory Pattern）在Python脚本中,特别适用于以下核心场景：

对象创建逻辑复杂

当创建对象需要：

多个配置参数组合
依赖外部配置或环境
需要执行校验或预处理

class DataLoaderFactory:
    @staticmethod
    def create_loader(source_type, config):
        if source_type == "csv":
            # 复杂的CSV加载器初始化
            return CSVLoader(
                delimiter=config.get('delimiter', ','),
                encoding=config.get('encoding', 'utf-8')
            )
        elif source_type == "json":
            # JSON加载器需要额外校验
            return JSONLoader(
                schema=config.get('schema'),
                strict=config.get('strict', False)
            )

需要解耦客户端和具体类

使用工厂模式避免脚本中直接硬编码具体类：

# 不推荐：客户端直接实例化
if format == "xml":
    report = XMLReportExporter("report.xml")
elif format == "pdf":
    report = PDFReportExporter("report.pdf")
# 推荐：通过工厂解耦
class ReportFactory:
    def get_exporter(self, format):
        exporters = {
            "xml": XMLReportExporter,
            "pdf": PDFReportExporter,
            "html": HTMLReportExporter
        }
        return exporters[format]("report." + format)

需要动态选择算法/策略

在数据分析脚本中,根据不同输入选择不同算法：

class DataProcessingFactory:
    @staticmethod
    def get_processor(data_type):
        processors = {
            "numeric": NumericProcessor(),
            "text": TextProcessor(),
            "time_series": TimeSeriesProcessor()
        }
        return processors.get(data_type, DefaultProcessor())

数据库/数据源切换

脚本需要连接不同类型的数据库或数据源：

class DatabaseConnectionFactory:
    def get_connection(self, db_type, credentials):
        if db_type == "postgresql":
            return psycopg2.connect(**credentials)
        elif db_type == "mongodb":
            return pymongo.MongoClient(**credentials)
        elif db_type == "sqlite":
            return sqlite3.connect(credentials['database'])

日志/监控系统集成

根据环境配置不同的日志处理器：

class LoggerFactory:
    def create_logger(self, env):
        if env == "production":
            return ProductionLogger(
                level="ERROR",
                handler=CloudLogHandler()
            )
        elif env == "development":
            return DevLogger(
                level="DEBUG",
                handler=ConsoleHandler()
            )

插件系统

在需要动态加载不同插件的脚本中：

class PluginFactory:
    _plugins = {}
    @classmethod
    def register(cls, plugin_type, plugin_class):
        cls._plugins[plugin_type] = plugin_class
    @classmethod
    def create_plugin(cls, plugin_type):
        return cls._plugins[plugin_type]()
# 使用示例
PluginFactory.register("email", EmailNotificationPlugin)
PluginFactory.register("sms", SMSNotificationPlugin)
notification = PluginFactory.create_plugin("email")

测试模拟

在单元测试中创建模拟对象：

class MockFactory:
    def create_mock(self, service_type, test_scenario):
        if service_type == "api":
            return MockAPIResponse(test_scenario)
        elif service_type == "database":
            return MockDatabaseQuery(test_scenario)

资源池管理

管理数据库连接池、线程池等：

class ConnectionPoolFactory:
    _pools = {}
    @classmethod
    def get_pool(cls, pool_type, size=5):
        if pool_type not in cls._pools:
            if pool_type == "thread":
                cls._pools[pool_type] = ThreadPoolExecutor(size)
            elif pool_type == "connection":
                cls._pools[pool_type] = conn.ConnectionPool(size)
        return cls._pools[pool_type]

不适合使用的场景：

简单对象创建：只有一个类且初始化简单时
性能敏感代码：工厂模式增加了一层间接调用
没有必要的变化：后续不可能增加新的产品类型

Python特有的简单替代方案：

对于简单场景,可以考虑：

字典映射：代替工厂类
__init_subclass__：自动注册子类
装饰器：动态注册创建函数

# 简单字典实现
processors = {
    "csv": CSVParser,
    "json": JSONParser,
    "xml": XMLParser
}
def parse_file(filename, format):
    parser = processors[format]()
    return parser.parse(filename)

工厂模式的价值在于将对象的创建和使用分离，当你的脚本满足上述场景时,使用工厂模式能提高代码的可维护性和扩展性。