Factory Method Pattern

1
2
3
4
5
6
7
8

# 함수형 팩토리 예시
def create_storage_factory(provider: str):
    factories = {
        'aws': lambda config: S3Storage(**config),
        'azure': lambda config: BlobStorage(**config),
        'gcp': lambda config: GCPStorage(**config)
    }
    return factories.get(provider)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

from abc import ABC, abstractmethod

# Product 인터페이스
class Logger(ABC):
    @abstractmethod
    def log(self, message: str):
        pass

# ConcreteProduct 1
class FileLogger(Logger):
    def log(self, message: str):
        print(f"[FileLogger] {message}")

# ConcreteProduct 2
class DatabaseLogger(Logger):
    def log(self, message: str):
        print(f"[DatabaseLogger] {message}")

# Creator (추상 팩토리 클래스)
class LoggerFactory(ABC):
    @abstractmethod
    def create_logger(self) -> Logger:
        pass

# ConcreteCreator 1
class FileLoggerFactory(LoggerFactory):
    def create_logger(self) -> Logger:
        return FileLogger()

# ConcreteCreator 2
class DatabaseLoggerFactory(LoggerFactory):
    def create_logger(self) -> Logger:
        return DatabaseLogger()

# 클라이언트 코드
def log_message(factory: LoggerFactory, msg: str):
    logger = factory.create_logger()
    logger.log(msg)

log_message(FileLoggerFactory(), "파일에 기록")
log_message(DatabaseLoggerFactory(), "DB에 기록")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class Document:
    def render(self):
        raise NotImplementedError()

class PDFDocument(Document):
    def render(self):
        print("Rendering PDF")

class WordDocument(Document):
    def render(self):
        print("Rendering Word")

# 단일 팩토리 클래스: 매개변수 기반 제품 결정
class DocumentFactory:
    def create_document(self, doc_type: str) -> Document:
        if doc_type == "PDF":
            return PDFDocument()
        elif doc_type == "Word":
            return WordDocument()
        else:
            raise ValueError(f"Unknown document type: {doc_type}")

# 클라이언트
factory = DocumentFactory()
doc = factory.create_document("PDF")
doc.render()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

from typing import Callable

class Shape:
    def draw(self):
        raise NotImplementedError()

class Circle(Shape):
    def draw(self):
        print("Circle drawn")

class Square(Shape):
    def draw(self):
        print("Square drawn")

# 등록 기반 팩토리
class ShapeFactory:
    def __init__(self):
        self._creators: dict[str, Callable[[], Shape]] = {}

    def register(self, shape_type: str, creator: Callable[[], Shape]):
        self._creators[shape_type] = creator

    def create(self, shape_type: str) -> Shape:
        creator = self._creators.get(shape_type)
        if not creator:
            raise ValueError(f"Unknown shape: {shape_type}")
        return creator()

# 팩토리 사용
factory = ShapeFactory()
factory.register("circle", Circle)
factory.register("square", Square)

shape = factory.create("square")
shape.draw()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class Connection:
    def __init__(self, db_type: str):
        self.db_type = db_type

    @staticmethod
    def create_mysql():
        return Connection("MySQL")

    @staticmethod
    def create_postgres():
        return Connection("PostgreSQL")

# 사용
conn1 = Connection.create_mysql()
print(f"Connected to {conn1.db_type}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from typing import Callable

class Notification:
    def send(self):
        raise NotImplementedError()

class EmailNotification(Notification):
    def send(self):
        print("Sending Email")

class SMSNotification(Notification):
    def send(self):
        print("Sending SMS")

# 함수형 팩토리 맵
notification_factory: dict[str, Callable[[], Notification]] = {
    "email": EmailNotification,
    "sms": SMSNotification,
}

# 클라이언트
def create_notification(ntype: str) -> Notification:
    return notification_factory[ntype]()

notif = create_notification("email")
notif.send()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 제품
class Service:
    def execute(self):
        print("Service executed")

# 팩토리 역할 수행 클래스
class ServiceFactory:
    def __init__(self, implementation: Service):
        self._impl = implementation

    def get_service(self) -> Service:
        return self._impl

# 클라이언트 코드: DI 주입 가정
factory = ServiceFactory(Service())
svc = factory.get_service()
svc.execute()

1

Client → GUIFactory → createButton() → WindowsButton or LinuxButton 생성 → UI 렌더링

1

Client → NotifierFactory → ConcreteFactory → Notifier (Email/Slack/SMS)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

from abc import ABC, abstractmethod

# 1. Product Interface
class Notifier(ABC):
    @abstractmethod
    def send(self, message: str):
        pass

# 2. Concrete Product
class EmailNotifier(Notifier):
    def send(self, message: str):
        print(f"[Email] {message}")

class SlackNotifier(Notifier):
    def send(self, message: str):
        print(f"[Slack] {message}")

class SMSNotifier(Notifier):
    def send(self, message: str):
        print(f"[SMS] {message}")

# 3. Creator (Factory)
class NotifierFactory(ABC):
    @abstractmethod
    def create_notifier(self) -> Notifier:
        pass

# 4. Concrete Factories
class EmailNotifierFactory(NotifierFactory):
    def create_notifier(self) -> Notifier:
        return EmailNotifier()

class SlackNotifierFactory(NotifierFactory):
    def create_notifier(self) -> Notifier:
        return SlackNotifier()

class SMSNotifierFactory(NotifierFactory):
    def create_notifier(self) -> Notifier:
        return SMSNotifier()

# 5. Client 코드
def notify(factory: NotifierFactory, message: str):
    notifier = factory.create_notifier()
    notifier.send(message)

# 사용 예시
if __name__ == "__main__":
    notify(EmailNotifierFactory(), "메일 전송 완료")
    notify(SlackNotifierFactory(), "슬랙 알림 전송")
    notify(SMSNotifierFactory(), "SMS 발송 성공")

1
2
3

[Client] → [PaymentFactory] → [CreditCardFactory] → [CreditCardPayment]
                                   ↑
                                   └→ [PaypalFactory] → [PaypalPayment]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

from abc import ABC, abstractmethod

# 1. Product 인터페이스
class Payment(ABC):
    @abstractmethod
    def pay(self, amount: float):
        pass

# 2. Concrete Products
class CreditCardPayment(Payment):
    def pay(self, amount: float):
        print(f"[CreditCard] {amount:f}원을 신용카드로 결제합니다.")

class PaypalPayment(Payment):
    def pay(self, amount: float):
        print(f"[PayPal] {amount:f}원을 PayPal로 결제합니다.")

class CryptoPayment(Payment):
    def pay(self, amount: float):
        print(f"[Crypto] {amount:f}원을 암호화폐로 결제합니다.")

# 3. Creator (Factory 인터페이스)
class PaymentFactory(ABC):
    @abstractmethod
    def create_payment(self) -> Payment:
        pass

# 4. Concrete Creators
class CreditCardFactory(PaymentFactory):
    def create_payment(self) -> Payment:
        return CreditCardPayment()

class PaypalFactory(PaymentFactory):
    def create_payment(self) -> Payment:
        return PaypalPayment()

class CryptoFactory(PaymentFactory):
    def create_payment(self) -> Payment:
        return CryptoPayment()

# 5. 클라이언트
def process_payment(factory: PaymentFactory, amount: float):
    payment = factory.create_payment()
    payment.pay(amount)

# 사용 예시
if __name__ == "__main__":
    process_payment(CreditCardFactory(), 10000)
    process_payment(PaypalFactory(), 20000)
    process_payment(CryptoFactory(), 5000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

from abc import ABC, abstractmethod

# Product
class DatabaseConnection(ABC):
    @abstractmethod
    def connect(self):
        pass

# Concrete Products
class MySQLConnection(DatabaseConnection):
    def connect(self):
        print("MySQL 연결 완료")

class OracleConnection(DatabaseConnection):
    def connect(self):
        print("Oracle 연결 완료")

# Creator
class DatabaseConnectionFactory(ABC):
    @abstractmethod
    def create_connection(self) -> DatabaseConnection:
        pass

# Concrete Creators
class MySQLConnectionFactory(DatabaseConnectionFactory):
    def create_connection(self):
        return MySQLConnection()

class OracleConnectionFactory(DatabaseConnectionFactory):
    def create_connection(self):
        return OracleConnection()

# Client
def client(factory: DatabaseConnectionFactory):
    conn = factory.create_connection()
    conn.connect()

client(MySQLConnectionFactory())
client(OracleConnectionFactory())

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105

from abc import ABC, abstractmethod
from typing import Dict, Any
import json
import yaml
import xml.etree.ElementTree as ET

# 추상 제품 클래스
class ConfigParser(ABC):
    @abstractmethod
    def parse(self, config_string: str) -> Dict[str, Any]:
        """설정 문자열을 파싱하여 딕셔너리로 반환합니다."""
        pass
    
    @abstractmethod
    def get_format(self) -> str:
        """파서의 형식을 반환합니다."""
        pass

# 구체적인 제품 클래스들
class JSONConfigParser(ConfigParser):
    def parse(self, config_string: str) -> Dict[str, Any]:
        try:
            return json.loads(config_string)
        except json.JSONDecodeError as e:
            raise ValueError(f"JSON 파싱 오류: {str(e)}")
            
    def get_format(self) -> str:
        return "JSON"

class YAMLConfigParser(ConfigParser):
    def parse(self, config_string: str) -> Dict[str, Any]:
        try:
            return yaml.safe_load(config_string)
        except yaml.YAMLError as e:
            raise ValueError(f"YAML 파싱 오류: {str(e)}")
            
    def get_format(self) -> str:
        return "YAML"

class XMLConfigParser(ConfigParser):
    def parse(self, config_string: str) -> Dict[str, Any]:
        try:
            root = ET.fromstring(config_string)
            return self._element_to_dict(root)
        except ET.ParseError as e:
            raise ValueError(f"XML 파싱 오류: {str(e)}")
            
    def get_format(self) -> str:
        return "XML"
        
    def _element_to_dict(self, element: ET.Element) -> Dict[str, Any]:
        result = {}
        for child in element:
            if len(child) == 0:
                result[child.tag] = child.text
            else:
                result[child.tag] = self._element_to_dict(child)
        return result

# 팩토리 클래스
class ConfigParserFactory:
    _parsers = {}
    
    @classmethod
    def register_parser(cls, format_type: str, parser_class: type) -> None:
        """새로운 파서를 등록합니다."""
        cls._parsers[format_type.lower()] = parser_class
        
    @classmethod
    def get_parser(cls, format_type: str) -> ConfigParser:
        """요청된 형식의 파서를 생성하여 반환합니다."""
        parser_class = cls._parsers.get(format_type.lower())
        if not parser_class:
            raise ValueError(f"지원하지 않는 형식입니다: {format_type}")
        return parser_class()

# 파서 등록
ConfigParserFactory.register_parser("json", JSONConfigParser)
ConfigParserFactory.register_parser("yaml", YAMLConfigParser)
ConfigParserFactory.register_parser("xml", XMLConfigParser)

# 사용 예시
def parse_config(format_type: str, config_string: str) -> Dict[str, Any]:
    try:
        parser = ConfigParserFactory.get_parser(format_type)
        print(f"{parser.get_format()} 파서를 사용하여 파싱을 시작합니다.")
        return parser.parse(config_string)
    except Exception as e:
        print(f"파싱 중 오류 발생: {str(e)}")
        raise

# 테스트
if __name__ == "__main__":
    # JSON 설정 파싱
    json_config = '{"name": "test", "value": 123}'
    result = parse_config("json", json_config)
    print("JSON 결과:", result)
    
    # YAML 설정 파싱
    yaml_config = """
    name: test
    value: 123
    """
    result = parse_config("yaml", yaml_config)
    print("YAML 결과:", result)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

// 추상 제품 클래스 (인터페이스)
class PaymentProcessor {
    process(amount) {
        throw new Error('process 메서드를 구현해야 합니다.');
    }
    
    getType() {
        throw new Error('getType 메서드를 구현해야 합니다.');
    }
}

// 구체적인 제품 클래스들
class CreditCardProcessor extends PaymentProcessor {
    constructor(apiKey) {
        super();
        this.apiKey = apiKey;
    }
    
    async process(amount) {
        console.log(`신용카드 결제 처리 중: $${amount}`);
        // 실제 신용카드 결제 처리 로직
        return {
            success: true,
            type: 'credit_card',
            amount,
            timestamp: new Date()
        };
    }
    
    getType() {
        return 'Credit Card';
    }
}

class PayPalProcessor extends PaymentProcessor {
    constructor(clientId, clientSecret) {
        super();
        this.clientId = clientId;
        this.clientSecret = clientSecret;
    }
    
    async process(amount) {
        console.log(`PayPal 결제 처리 중: $${amount}`);
        // 실제 PayPal 결제 처리 로직
        return {
            success: true,
            type: 'paypal',
            amount,
            timestamp: new Date()
        };
    }
    
    getType() {
        return 'PayPal';
    }
}

// 팩토리 클래스
class PaymentProcessorFactory {
    static _processors = new Map();
    static _configs = new Map();
    
    static registerProcessor(type, processorClass, config) {
        this._processors.set(type.toLowerCase(), processorClass);
        if (config) {
            this._configs.set(type.toLowerCase(), config);
        }
    }
    
    static createProcessor(type) {
        const processorClass = this._processors.get(type.toLowerCase());
        if (!processorClass) {
            throw new Error(`지원하지 않는 결제 방식입니다: ${type}`);
        }
        
        const config = this._configs.get(type.toLowerCase());
        return new processorClass(…Object.values(config));
    }
}

// 결제 프로세서 등록
PaymentProcessorFactory.registerProcessor('credit_card', CreditCardProcessor, {
    apiKey: 'test_api_key'
});

PaymentProcessorFactory.registerProcessor('paypal', PayPalProcessor, {
    clientId: 'test_client_id',
    clientSecret: 'test_client_secret'
});

// 결제 처리 서비스
class PaymentService {
    async processPayment(type, amount) {
        try {
            const processor = PaymentProcessorFactory.createProcessor(type);
            console.log(`${processor.getType()} 프로세서를 사용하여 결제를 시작합니다.`);
            return await processor.process(amount);
        } catch (error) {
            console.error('결제 처리 중 오류 발생:', error.message);
            throw error;
        }
    }
}

// 사용 예시
async function main() {
    const paymentService = new PaymentService();
    
    try {
        // 신용카드 결제
        const creditCardResult = await paymentService.processPayment('credit_card', 100);
        console.log('신용카드 결제 결과:', creditCardResult);
        
        // PayPal 결제
        const paypalResult = await paymentService.processPayment('paypal', 150);
        console.log('PayPal 결제 결과:', paypalResult);
    } catch (error) {
        console.error('결제 처리 실패:', error.message);
    }
}

main();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

interface PaymentProcessor {
    void process();
}

class RealPaymentProcessor implements PaymentProcessor {
    public void process() { /* 실제 결제 */ }
}

class PaymentProcessorFactory {
    public PaymentProcessor create() {
        return new RealPaymentProcessor();
    }
}

@Test
void testPaymentProcessing() {
    PaymentProcessor mock = Mockito.mock(PaymentProcessor.class);
    PaymentProcessorFactory testFactory = Mockito.mock(PaymentProcessorFactory.class);
    Mockito.when(testFactory.create()).thenReturn(mock);

    // 테스트 대상에 팩토리 주입
    OrderService service = new OrderService(testFactory);
    service.checkout();

    Mockito.verify(mock).process();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 단순 문자열 포장을 위해 과하게 팩토리 적용
class StringWrapper:
    def __init__(self, text: str):
        self.text = text

class StringFactory:
    def create_string(self, text: str) -> StringWrapper:
        return StringWrapper(text)

s = StringFactory().create_string("hello")
print(s.text)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 단순한 객체는 팩토리 없이 생성자 호출 또는 헬퍼 함수 사용
def create_string_wrapper(text: str) -> 'StringWrapper':
    return StringWrapper(text)

class StringWrapper:
    def __init__(self, text: str):
        self.text = text

s = create_string_wrapper("hello")
print(s.text)

1
2
3
4
5
6
7
8

# Product 인터페이스가 너무 구체적
class Animal:
    def walk_on_four_legs(self):  # 모든 동물이 네 발로 걷지는 않음
        pass

class Bird(Animal):
    def walk_on_four_legs(self):
        raise NotImplementedError("Birds don't walk on four legs")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from abc import ABC, abstractmethod

class Animal(ABC):
    @abstractmethod
    def move(self):  # 더 일반적인 행위로 추상화
        pass

class Dog(Animal):
    def move(self):
        print("Dog walks on four legs")

class Bird(Animal):
    def move(self):
        print("Bird flies")

def describe(animal: Animal):
    animal.move()

describe(Dog())
describe(Bird())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class Logger:
    def log(self, message: str):
        print(f"[LOG] {message}")

class LoggerFactory:
    def create_logger(self):
        logger = Logger()
        print("Logger created")  # 부수효과
        logger.log("Logger is ready")  # 부수효과
        return logger

logger = LoggerFactory().create_logger()
logger.log("작업 시작")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class Logger:
    def log(self, message: str):
        print(f"[LOG] {message}")

class LoggerFactory:
    def create_logger(self) -> Logger:
        return Logger()  # 생성만 수행

# 생성과 초기화 책임 분리
logger = LoggerFactory().create_logger()
print("Logger created")
logger.log("Logger is ready")
logger.log("작업 시작")