Python面向对象设计——工厂模式

一、什么是工厂模式？

工厂模式（Factory Pattern） 是一种创建型设计模式，它提供了一种创建对象的方式，而不需要指定具体的类。工厂模式将对象的创建与使用分离，通过一个共同的接口来创建对象，使得系统在不修改现有代码的情况下可以扩展新的产品类型。

# 不使用工厂模式
classDog:
defspeak(self):
return"汪汪"

classCat:
defspeak(self):
return"喵喵"

# 客户端代码直接实例化具体类
animal = Dog()
print(animal.speak())

# 使用工厂模式
classAnimalFactory:
defcreate_animal(self, animal_type):
if animal_type == "dog":
return Dog()
elif animal_type == "cat":
return Cat()

factory = AnimalFactory()
animal = factory.create_animal("dog")
print(animal.speak())

二、为什么需要工厂模式？

2.1 适用场景

• • 创建逻辑复杂：对象的创建涉及多个步骤或依赖其他对象
• • 需要解耦：客户端不需要知道具体的产品类
• • 需要扩展性：新增产品类型时不需要修改客户端代码
• • 创建条件不确定：根据运行时条件决定创建哪种对象

2.2 优点

• • 解耦对象创建和使用，降低系统耦合度
• • 符合开闭原则，扩展新产品不需要修改客户端
• • 集中管理对象创建逻辑，便于维护
• • 隐藏复杂的创建细节

2.3 缺点

• • 增加代码复杂度，需要引入额外的类和接口
• • 对于简单的对象创建，可能过度设计

三、工厂模式的类型

四、简单工厂模式

4.1 概念

简单工厂不是正式的设计模式，而是一种编程习惯。它通过一个工厂类根据传入的参数决定创建哪种产品。

from abc import ABC, abstractmethod

# 产品接口
classAnimal(ABC):
    @abstractmethod
defspeak(self):
pass

# 具体产品
classDog(Animal):
defspeak(self):
return"汪汪"

classCat(Animal):
defspeak(self):
return"喵喵"

classBird(Animal):
defspeak(self):
return"叽叽喳喳"

# 简单工厂
classAnimalFactory:
    @staticmethod
defcreate_animal(animal_type: str) -> Animal:
if animal_type == "dog":
return Dog()
elif animal_type == "cat":
return Cat()
elif animal_type == "bird":
return Bird()
else:
raise ValueError(f"未知的动物类型: {animal_type}")

# 使用
factory = AnimalFactory()
animal = factory.create_animal("dog")
print(animal.speak())  # 汪汪

4.2 使用类方法实现简单工厂

classAnimal(ABC):
    @abstractmethod
defspeak(self):
pass

    @classmethod
defcreate(cls, animal_type: str):
if animal_type == "dog":
return Dog()
elif animal_type == "cat":
return Cat()
else:
raise ValueError(f"未知类型: {animal_type}")

classDog(Animal):
defspeak(self):
return"汪汪"

# 使用
animal = Animal.create("dog")
print(animal.speak())

五、工厂方法模式

5.1 概念

工厂方法模式定义了一个创建对象的接口，但由子类决定实例化哪一个类。工厂方法让类的实例化延迟到子类中。

from abc import ABC, abstractmethod

# 产品接口
classTransport(ABC):
    @abstractmethod
defdeliver(self):
pass

# 具体产品
classTruck(Transport):
defdeliver(self):
return"陆路运输（卡车）"

classShip(Transport):
defdeliver(self):
return"海上运输（船只）"

classAirplane(Transport):
defdeliver(self):
return"空中运输（飞机）"

# 工厂抽象
classLogistics(ABC):
    @abstractmethod
defcreate_transport(self) -> Transport:
pass

defplan_delivery(self):
        transport = self.create_transport()
return transport.deliver()

# 具体工厂
classRoadLogistics(Logistics):
defcreate_transport(self) -> Transport:
return Truck()

classSeaLogistics(Logistics):
defcreate_transport(self) -> Transport:
return Ship()

classAirLogistics(Logistics):
defcreate_transport(self) -> Transport:
return Airplane()

# 使用
logistics = RoadLogistics()
print(logistics.plan_delivery())  # 陆路运输（卡车）

logistics = SeaLogistics()
print(logistics.plan_delivery())  # 海上运输（船只）

logistics = AirLogistics()
print(logistics.plan_delivery())  # 空中运输（飞机）

六、抽象工厂模式

6.1 概念

抽象工厂模式创建一系列相关或相互依赖的产品族，而不需要指定具体的类。

from abc import ABC, abstractmethod

# 产品接口 - 椅子
classChair(ABC):
    @abstractmethod
defsit_on(self):
pass

# 产品接口 - 桌子
classTable(ABC):
    @abstractmethod
defput_on(self):
pass

# 产品接口 - 沙发
classSofa(ABC):
    @abstractmethod
deflie_on(self):
pass

# 现代风格产品
classModernChair(Chair):
defsit_on(self):
return"坐在现代椅子上"

classModernTable(Table):
defput_on(self):
return"现代桌子"

classModernSofa(Sofa):
deflie_on(self):
return"现代沙发"

# 古典风格产品
classVintageChair(Chair):
defsit_on(self):
return"坐在古典椅子上"

classVintageTable(Table):
defput_on(self):
return"古典桌子"

classVintageSofa(Sofa):
deflie_on(self):
return"古典沙发"

# 抽象工厂
classFurnitureFactory(ABC):
    @abstractmethod
defcreate_chair(self) -> Chair:
pass

    @abstractmethod
defcreate_table(self) -> Table:
pass

    @abstractmethod
defcreate_sofa(self) -> Sofa:
pass

# 现代风格工厂
classModernFurnitureFactory(FurnitureFactory):
defcreate_chair(self) -> Chair:
return ModernChair()

defcreate_table(self) -> Table:
return ModernTable()

defcreate_sofa(self) -> Sofa:
return ModernSofa()

# 古典风格工厂
classVintageFurnitureFactory(FurnitureFactory):
defcreate_chair(self) -> Chair:
return VintageChair()

defcreate_table(self) -> Table:
return VintageTable()

defcreate_sofa(self) -> Sofa:
return VintageSofa()

# 使用
defdecorate_room(factory: FurnitureFactory):
    chair = factory.create_chair()
    table = factory.create_table()
    sofa = factory.create_sofa()

print(chair.sit_on())
print(table.put_on())
print(sofa.lie_on())

print("=== 现代风格房间 ===")
decorate_room(ModernFurnitureFactory())

print("\n=== 古典风格房间 ===")
decorate_room(VintageFurnitureFactory())

七、实战案例

7.1 数据库连接工厂

from abc import ABC, abstractmethod
import sqlite3
import pymysql  # 需要安装
import psycopg2  # 需要安装

# 产品接口
classDatabaseConnection(ABC):
    @abstractmethod
defconnect(self):
pass

    @abstractmethod
defexecute(self, query):
pass

# 具体产品
classSQLiteConnection(DatabaseConnection):
defconnect(self):
self.conn = sqlite3.connect('database.db')
returnself.conn

defexecute(self, query):
        cursor = self.conn.cursor()
        cursor.execute(query)
return cursor.fetchall()

classMySQLConnection(DatabaseConnection):
defconnect(self):
self.conn = pymysql.connect(
            host='localhost',
            user='root',
            password='password',
            database='test'
        )
returnself.conn

defexecute(self, query):
        cursor = self.conn.cursor()
        cursor.execute(query)
return cursor.fetchall()

classPostgreSQLConnection(DatabaseConnection):
defconnect(self):
self.conn = psycopg2.connect(
            host='localhost',
            user='postgres',
            password='password',
            database='test'
        )
returnself.conn

defexecute(self, query):
        cursor = self.conn.cursor()
        cursor.execute(query)
return cursor.fetchall()

# 工厂
classDatabaseFactory:
    @staticmethod
defcreate_connection(db_type: str) -> DatabaseConnection:
if db_type == 'sqlite':
return SQLiteConnection()
elif db_type == 'mysql':
return MySQLConnection()
elif db_type == 'postgres':
return PostgreSQLConnection()
else:
raise ValueError(f"不支持的数据库类型: {db_type}")

# 使用
db = DatabaseFactory.create_connection('sqlite')
db.connect()
# db.execute("SELECT * FROM users")

7.2 文档生成器工厂

from abc import ABC, abstractmethod

# 产品接口
classDocument(ABC):
    @abstractmethod
defcreate_header(self):
pass

    @abstractmethod
defcreate_body(self):
pass

    @abstractmethod
defcreate_footer(self):
pass

# 具体产品
classPDFDocument(Document):
defcreate_header(self):
return"PDF 页眉"

defcreate_body(self):
return"PDF 正文"

defcreate_footer(self):
return"PDF 页脚"

classHTMLDocument(Document):
defcreate_header(self):
return"<header>HTML Header</header>"

defcreate_body(self):
return"<body>HTML Body</body>"

defcreate_footer(self):
return"<footer>HTML Footer</footer>"

classMarkdownDocument(Document):
defcreate_header(self):
return"# Markdown Header"

defcreate_body(self):
return"## Body\n\nContent here"

defcreate_footer(self):
return"*Footer*"

# 工厂
classDocumentFactory:
    @staticmethod
defcreate_document(doc_type: str) -> Document:
if doc_type == 'pdf':
return PDFDocument()
elif doc_type == 'html':
return HTMLDocument()
elif doc_type == 'md':
return MarkdownDocument()
else:
raise ValueError(f"不支持的文档类型: {doc_type}")

defgenerate_report(doc_type: str):
    doc = DocumentFactory.create_document(doc_type)
    content = f"{doc.create_header()}\n{doc.create_body()}\n{doc.create_footer()}"
return content

print("=== PDF 报告 ===")
print(generate_report('pdf'))

print("\n=== HTML 报告 ===")
print(generate_report('html'))

print("\n=== Markdown 报告 ===")
print(generate_report('md'))

7.3 支付网关工厂（工厂方法）

from abc import ABC, abstractmethod

# 产品接口
classPayment(ABC):
    @abstractmethod
defpay(self, amount: float) -> bool:
pass

    @abstractmethod
defrefund(self, amount: float) -> bool:
pass

# 具体产品
classAlipayPayment(Payment):
def__init__(self, app_id: str):
self.app_id = app_id

defpay(self, amount: float) -> bool:
print(f"支付宝支付 {amount} 元")
returnTrue

defrefund(self, amount: float) -> bool:
print(f"支付宝退款 {amount} 元")
returnTrue

classWechatPayment(Payment):
def__init__(self, mch_id: str):
self.mch_id = mch_id

defpay(self, amount: float) -> bool:
print(f"微信支付 {amount} 元")
returnTrue

defrefund(self, amount: float) -> bool:
print(f"微信退款 {amount} 元")
returnTrue

classCreditCardPayment(Payment):
def__init__(self, card_number: str):
self.card_number = card_number

defpay(self, amount: float) -> bool:
print(f"信用卡支付 {amount} 元")
returnTrue

defrefund(self, amount: float) -> bool:
print(f"信用卡退款 {amount} 元")
returnTrue

# 工厂接口
classPaymentFactory(ABC):
    @abstractmethod
defcreate_payment(self) -> Payment:
pass

# 具体工厂
classAlipayFactory(PaymentFactory):
def__init__(self, app_id: str):
self.app_id = app_id

defcreate_payment(self) -> Payment:
return AlipayPayment(self.app_id)

classWechatFactory(PaymentFactory):
def__init__(self, mch_id: str):
self.mch_id = mch_id

defcreate_payment(self) -> Payment:
return WechatPayment(self.mch_id)

classCreditCardFactory(PaymentFactory):
def__init__(self, card_number: str):
self.card_number = card_number

defcreate_payment(self) -> Payment:
return CreditCardPayment(self.card_number)

# 使用
classPaymentProcessor:
def__init__(self, factory: PaymentFactory):
self.factory = factory

defprocess_payment(self, amount: float):
        payment = self.factory.create_payment()
return payment.pay(amount)

# 客户端
alipay_processor = PaymentProcessor(AlipayFactory("app_123456"))
alipay_processor.process_payment(100)

wechat_processor = PaymentProcessor(WechatFactory("mch_123456"))
wechat_processor.process_payment(200)

7.4 配置化工厂（动态注册）

from abc import ABC, abstractmethod
from typing importDict, Type

# 产品接口
classPlugin(ABC):
    @abstractmethod
defexecute(self, data):
pass

# 具体产品
classDataValidator(Plugin):
defexecute(self, data):
returnf"验证数据: {data}"

classDataTransformer(Plugin):
defexecute(self, data):
returnf"转换数据: {data}"

classDataExporter(Plugin):
defexecute(self, data):
returnf"导出数据: {data}"

# 可扩展工厂（支持动态注册）
classPluginFactory:
    _plugins: Dict[str, Type[Plugin]] = {}

    @classmethod
defregister(cls, name: str, plugin_class: Type[Plugin]):
        cls._plugins[name] = plugin_class

    @classmethod
defcreate(cls, name: str, **kwargs) -> Plugin:
        plugin_class = cls._plugins.get(name)
ifnot plugin_class:
raise ValueError(f"未知的插件: {name}")
return plugin_class(**kwargs)

    @classmethod
deflist_plugins(cls):
returnlist(cls._plugins.keys())

# 注册插件
PluginFactory.register("validator", DataValidator)
PluginFactory.register("transformer", DataTransformer)
PluginFactory.register("exporter", DataExporter)

# 使用
plugin = PluginFactory.create("validator")
print(plugin.execute("测试数据"))

plugin = PluginFactory.create("transformer")
print(plugin.execute("测试数据"))

print(f"可用插件: {PluginFactory.list_plugins()}")

八、工厂模式对比

九、总结

核心要点：

• • 简单工厂：一个工厂类根据参数创建不同产品，适合产品类型较少的场景。
• • 工厂方法：定义创建接口，子类决定创建哪种产品，符合开闭原则。
• • 抽象工厂：创建一系列相关产品族，适合需要统一风格或关联产品的场景。
• • 优点：解耦创建和使用、符合开闭原则、集中管理创建逻辑。
• • 缺点：增加代码复杂度，需要引入额外类和接口。

选择建议：

• • 产品类型少且固定 → 简单工厂
• • 产品类型多且需要扩展 → 工厂方法
• • 需要创建一系列相关产品 → 抽象工厂

掌握工厂模式，可以帮助你在对象创建逻辑复杂或需要支持扩展时，写出更灵活、更易维护的代码。