Factory Design Pattern

🔍 1단계: 추상 인터페이스 정의
코드
python
복사
편집
from abc import ABC, abstractmethod

class Coffee(ABC):
    @abstractmethod
    def prepare(self):
        pass
설명:
**추상 클래스(인터페이스)**인 Coffee를 정의한다.

Coffee 클래스는 직접 인스턴스를 생성할 수 없으며, 반드시 하위 클래스가 prepare() 메서드를 구현하도록 강제한다.

이를 통해 커피 종류가 달라져도 항상 같은 메서드(prepare)로 준비 과정을 처리할 수 있다.

☕️ 2단계: Concrete Product(구체적인 제품 클래스) 구현
코드
python
복사
편집
class Espresso(Coffee):
    def prepare(self):
        return "Preparing a rich and strong Espresso."


class Latte(Coffee):
    def prepare(self):
        return "Preparing a smooth and creamy Latte."


class Cappuccino(Coffee):
    def prepare(self):
        return "Preparing a frothy Cappuccino."
설명:
Concrete Product 클래스들(Espresso, Latte, Cappuccino)이 추상 클래스 Coffee를 상속하여 각기 다른 prepare 메서드를 구체적으로 구현한다.

각각의 클래스가 구체적인 제품의 제조법(메시지)을 정의한다.

🏭 3단계: Factory 클래스 구현 (CoffeeMachine)
코드
python
복사
편집
class CoffeeMachine:
    def make_coffee(self, coffee_type):
        if coffee_type == "Espresso":
            return Espresso().prepare()
        elif coffee_type == "Latte":
            return Latte().prepare()
        elif coffee_type == "Cappuccino":
            return Cappuccino().prepare()
        else:
            return "Unknown coffee type!"
설명:
Factory 역할을 하는 클래스(CoffeeMachine)를 통해 구체적인 커피 객체를 생성하고 준비 과정(prepare())을 수행한다.

사용자는 커피 타입(coffee_type)만 지정하면, Factory에서 조건을 판단하여 적절한 Concrete Product를 생성 및 준비한다.

이렇게 되면 사용자는 직접 커피 객체를 생성하지 않아도 된다. 즉, 커피 제작 방식의 변화나 새로운 커피 종류 추가 시 Factory만 수정하면 되어 유지보수성이 높아진다.

🎯 4단계: Factory 사용 (메인 프로그램)
코드
python
복사
편집
if __name__ == "__main__":
    machine = CoffeeMachine()

    coffee = machine.make_coffee("Espresso")
    print(coffee)  # Preparing a rich and strong Espresso.

    coffee = machine.make_coffee("Latte")
    print(coffee)  # Preparing a smooth and creamy Latte.

    coffee = machine.make_coffee("Cappuccino")
    print(coffee)  # Preparing a frothy Cappuccino.
설명:
메인 프로그램에서 CoffeeMachine 인스턴스를 만들고 make_coffee() 메서드를 호출하여 원하는 커피를 주문한다.

이 방식은 클라이언트가 복잡한 객체 생성 과정을 알지 않아도 간단한 인터페이스로 원하는 객체를 얻을 수 있게 한다.

📝 패턴 요약 (Factory Method Pattern)
Coffee (추상 클래스) → 추상 제품 (Product)

Espresso, Latte, Cappuccino → 구체적인 제품 (Concrete Product)

CoffeeMachine → Factory 클래스: 제품의 인스턴스를 생성하는 공장 역할

이렇게 하면 제품 객체 생성 과정을 분리하여, 코드의 확장성과 유지보수성을 크게 높일 수 있다.

즉, 새로운 커피 타입(예: Americano, Mocha)이 추가될 경우:

Concrete Product를 새로 구현하고,

Factory 메서드에 조건문만 추가해주면 된다.

이로써 코드가 더욱 깔끔하고 유연하게 관리될 수 있다.

 

 

활용 예시 - 학습에 사용할 방대한 양의 데이터가 다양한 형식을 가지고 있을 때에 대한 처리

import os
import zipfile
from abc import ABC, abstractmethod

import pandas as pd


# Define an abstract class for Data Ingestor
class DataIngestor(ABC):
    @abstractmethod
    def ingest(self, file_path: str) -> pd.DataFrame:
        """Abstract method to ingest data from a given file."""
        pass


# Implement a concrete class for ZIP Ingestion
class ZipDataIngestor(DataIngestor):
    def ingest(self, file_path: str) -> pd.DataFrame:
        """Extracts a .zip file and returns the content as a pandas DataFrame."""
        # Ensure the file is a .zip
        if not file_path.endswith(".zip"):
            raise ValueError("The provided file is not a .zip file.")

        # Extract the zip file
        with zipfile.ZipFile(file_path, "r") as zip_ref:
            zip_ref.extractall("extracted_data")

        # Find the extracted CSV file (assuming there is one CSV file inside the zip)
        extracted_files = os.listdir("extracted_data")
        csv_files = [f for f in extracted_files if f.endswith(".csv")]

        if len(csv_files) == 0:
            raise FileNotFoundError("No CSV file found in the extracted data.")
        if len(csv_files) > 1:
            raise ValueError("Multiple CSV files found. Please specify which one to use.")

        # Read the CSV into a DataFrame
        csv_file_path = os.path.join("extracted_data", csv_files[0])
        df = pd.read_csv(csv_file_path)

        # Return the DataFrame
        return df


# Implement a Factory to create DataIngestors
class DataIngestorFactory:
    @staticmethod
    def get_data_ingestor(file_extension: str) -> DataIngestor:
        """Returns the appropriate DataIngestor based on file extension."""
        if file_extension == ".zip":
            return ZipDataIngestor()
        else:
            raise ValueError(f"No ingestor available for file extension: {file_extension}")


# Example usage:
if __name__ == "__main__":
    # # Specify the file path
    # file_path = "/Users/ayushsingh/Desktop/end-to-end-production-grade-projects/prices-predictor-system/data/archive.zip"

    # # Determine the file extension
    # file_extension = os.path.splitext(file_path)[1]

    # # Get the appropriate DataIngestor
    # data_ingestor = DataIngestorFactory.get_data_ingestor(file_extension)

    # # Ingest the data and load it into a DataFrame
    # df = data_ingestor.ingest(file_path)

    # # Now df contains the DataFrame from the extracted CSV
    # print(df.head())  # Display the first few rows of the DataFrame
    pass