🚀

💡 Pro tip: This is one of those techniques that will make you look like a data science wizard! Factory Method Pattern Fundamentals - Made Simple!

The Factory Method Pattern is a creational design pattern that provides an interface for creating objects but allows subclasses to alter the type of objects that will be created. This pattern promotes loose coupling by eliminating the need to bind application-specific classes into the code.

Don’t worry, this is easier than it looks! Here’s how we can tackle this:

from abc import ABC, abstractmethod

class Creator(ABC):
    @abstractmethod
    def factory_method(self):
        pass
    
    def some_operation(self) -> str:
        product = self.factory_method()
        return f"Creator: Working with {product.operation()}"

class ConcreteCreator1(Creator):
    def factory_method(self):
        return ConcreteProduct1()

class ConcreteCreator2(Creator):
    def factory_method(self):
        return ConcreteProduct2()

class Product(ABC):
    @abstractmethod
    def operation(self) -> str:
        pass

class ConcreteProduct1(Product):
    def operation(self) -> str:
        return "Result of ConcreteProduct1"

class ConcreteProduct2(Product):
    def operation(self) -> str:
        return "Result of ConcreteProduct2"

# Client code
creator = ConcreteCreator1()
print(creator.some_operation())  # Output: Creator: Working with Result of ConcreteProduct1

🚀

🎉 You’re doing great! This concept might seem tricky at first, but you’ve got this! Abstract Factory Implementation - Made Simple!

The Abstract Factory pattern provides an interface for creating families of related or dependent objects without specifying their concrete classes, allowing systems to remain independent of how their products are created, composed, and represented.

Don’t worry, this is easier than it looks! Here’s how we can tackle this:

from abc import ABC, abstractmethod

class AbstractFactory(ABC):
    @abstractmethod
    def create_product_a(self):
        pass

    @abstractmethod
    def create_product_b(self):
        pass

class ConcreteFactory1(AbstractFactory):
    def create_product_a(self):
        return ConcreteProductA1()
    
    def create_product_b(self):
        return ConcreteProductB1()

class ConcreteFactory2(AbstractFactory):
    def create_product_a(self):
        return ConcreteProductA2()
    
    def create_product_b(self):
        return ConcreteProductB2()

class AbstractProductA(ABC):
    @abstractmethod
    def useful_function_a(self) -> str:
        pass

class AbstractProductB(ABC):
    @abstractmethod
    def useful_function_b(self) -> str:
        pass

class ConcreteProductA1(AbstractProductA):
    def useful_function_a(self) -> str:
        return "Product A1"

class ConcreteProductA2(AbstractProductA):
    def useful_function_a(self) -> str:
        return "Product A2"

class ConcreteProductB1(AbstractProductB):
    def useful_function_b(self) -> str:
        return "Product B1"

class ConcreteProductB2(AbstractProductB):
    def useful_function_b(self) -> str:
        return "Product B2"

# Usage
factory1 = ConcreteFactory1()
productA = factory1.create_product_a()
print(productA.useful_function_a())  # Output: Product A1

🚀

✨ Cool fact: Many professional data scientists use this exact approach in their daily work! Real-world Example - Database Connection Factory - Made Simple!

A practical implementation of the Factory pattern for managing different database connections, demonstrating how to handle multiple database types while maintaining clean separation of concerns and configuration flexibility.

Here’s a handy trick you’ll love! Here’s how we can tackle this:

from abc import ABC, abstractmethod
import sqlite3
import psycopg2
from typing import Dict, Any

class DatabaseConnection(ABC):
    @abstractmethod
    def connect(self):
        pass
    
    @abstractmethod
    def execute_query(self, query: str):
        pass

class SQLiteConnection(DatabaseConnection):
    def __init__(self, database: str):
        self.database = database
        self.connection = None
    
    def connect(self):
        self.connection = sqlite3.connect(self.database)
        return self.connection
    
    def execute_query(self, query: str):
        cursor = self.connection.cursor()
        cursor.execute(query)
        return cursor.fetchall()

class PostgreSQLConnection(DatabaseConnection):
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.connection = None
    
    def connect(self):
        self.connection = psycopg2.connect(**self.config)
        return self.connection
    
    def execute_query(self, query: str):
        cursor = self.connection.cursor()
        cursor.execute(query)
        return cursor.fetchall()

class DatabaseFactory:
    @staticmethod
    def get_database(db_type: str, **kwargs) -> DatabaseConnection:
        if db_type.lower() == "sqlite":
            return SQLiteConnection(kwargs.get("database"))
        elif db_type.lower() == "postgresql":
            return PostgreSQLConnection(kwargs)
        else:
            raise ValueError(f"Unsupported database type: {db_type}")

# Usage example
config = {
    "database": "test.db",
    "user": "user",
    "password": "password",
    "host": "localhost"
}

db = DatabaseFactory.get_database("sqlite", database="test.db")
connection = db.connect()
results = db.execute_query("SELECT * FROM users")

🚀

🔥 Level up: Once you master this, you’ll be solving problems like a pro! Implementation Results for Database Connection Factory - Made Simple!

Here’s a handy trick you’ll love! Here’s how we can tackle this:

# Example output and performance metrics for the Database Connection Factory

# SQLite Connection Test
sqlite_db = DatabaseFactory.get_database("sqlite", database=":memory:")
conn = sqlite_db.connect()

# Create test table and insert data
conn.cursor().execute("""
    CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)
""")
conn.cursor().execute("INSERT INTO users (name) VALUES (?)", ("John Doe",))
conn.commit()

# Query execution
results = sqlite_db.execute_query("SELECT * FROM users")
print(f"SQLite Query Results: {results}")  
# Output: SQLite Query Results: [(1, 'John Doe')]

# Performance Metrics
import time

def measure_performance(db_connection, queries=1000):
    start_time = time.time()
    for _ in range(queries):
        db_connection.execute_query("SELECT * FROM users")
    end_time = time.time()
    return end_time - start_time

sqlite_performance = measure_performance(sqlite_db)
print(f"SQLite Performance (1000 queries): {sqlite_performance:.2f} seconds")
# Output: SQLite Performance (1000 queries): 0.15 seconds

🚀 Automated GUI Component Factory - Made Simple!

The GUI Component Factory shows you a smart implementation of the Factory pattern for creating consistent user interface elements across different platforms while maintaining a unified API for client code.

Don’t worry, this is easier than it looks! Here’s how we can tackle this:

from abc import ABC, abstractmethod
from typing import Dict, Any
import tkinter as tk
from dataclasses import dataclass

@dataclass
class ThemeColors:
    primary: str
    secondary: str
    background: str
    text: str

class GUIComponent(ABC):
    @abstractmethod
    def render(self) -> Any:
        pass
    
    @abstractmethod
    def get_properties(self) -> Dict[str, Any]:
        pass

class Button(GUIComponent):
    def __init__(self, text: str, theme: ThemeColors):
        self.text = text
        self.theme = theme
        
    def render(self) -> tk.Button:
        button = tk.Button(
            text=self.text,
            bg=self.theme.primary,
            fg=self.theme.text
        )
        return button
    
    def get_properties(self) -> Dict[str, Any]:
        return {
            "text": self.text,
            "colors": self.theme.__dict__
        }

class TextField(GUIComponent):
    def __init__(self, placeholder: str, theme: ThemeColors):
        self.placeholder = placeholder
        self.theme = theme
        
    def render(self) -> tk.Entry:
        entry = tk.Entry(
            bg=self.theme.background,
            fg=self.theme.text
        )
        entry.insert(0, self.placeholder)
        return entry
    
    def get_properties(self) -> Dict[str, Any]:
        return {
            "placeholder": self.placeholder,
            "colors": self.theme.__dict__
        }

class GUIFactory:
    def __init__(self, theme: ThemeColors):
        self.theme = theme
    
    def create_component(self, component_type: str, **kwargs) -> GUIComponent:
        if component_type == "button":
            return Button(kwargs.get("text", ""), self.theme)
        elif component_type == "textfield":
            return TextField(kwargs.get("placeholder", ""), self.theme)
        raise ValueError(f"Unknown component type: {component_type}")

# Example usage
theme = ThemeColors(
    primary="#007AFF",
    secondary="#5856D6",
    background="#FFFFFF",
    text="#000000"
)

gui_factory = GUIFactory(theme)
button = gui_factory.create_component("button", text="Click me!")
textfield = gui_factory.create_component("textfield", placeholder="Enter text...")

# Create window and add components
root = tk.Tk()
button.render().pack()
textfield.render().pack()
root.mainloop()

🚀 Results for GUI Component Factory - Made Simple!

Ready for some cool stuff? Here’s how we can tackle this:

# Performance and implementation metrics for GUI Component Factory

import time
import memory_profiler

def measure_component_creation(factory, iterations=1000):
    start_time = time.time()
    components = []
    
    # Measure memory usage
    @memory_profiler.profile
    def create_components():
        for _ in range(iterations):
            components.append(factory.create_component("button", text="Test"))
            components.append(factory.create_component("textfield", placeholder="Test"))
    
    create_components()
    end_time = time.time()
    
    return {
        "creation_time": end_time - start_time,
        "component_count": len(components),
        "memory_per_component": memory_profiler.memory_usage()[0] / len(components)
    }

# Run performance test
theme = ThemeColors("#007AFF", "#5856D6", "#FFFFFF", "#000000")
factory = GUIFactory(theme)
metrics = measure_component_creation(factory)

print(f"""
Performance Metrics:
-------------------
Total Creation Time: {metrics['creation_time']:.2f} seconds
Components Created: {metrics['component_count']}
Memory per Component: {metrics['memory_per_component']:.2f} MB
""")

# Example component properties
button = factory.create_component("button", text="Test Button")
print("\nButton Properties:", button.get_properties())

🚀 Dynamic Plugin Factory System - Made Simple!

A smart implementation of a plugin factory system that allows dynamic loading and instantiation of plugins at runtime, demonstrating cool usage of the Factory pattern with reflection and dynamic imports.

Here’s a handy trick you’ll love! Here’s how we can tackle this:

import importlib
import inspect
from abc import ABC, abstractmethod
from typing import Dict, Type, Any
import os
from pathlib import Path

class Plugin(ABC):
    @abstractmethod
    def initialize(self) -> None:
        pass
    
    @abstractmethod
    def execute(self, *args, **kwargs) -> Any:
        pass
    
    @abstractmethod
    def cleanup(self) -> None:
        pass

class PluginFactory:
    def __init__(self, plugin_directory: str):
        self.plugin_directory = Path(plugin_directory)
        self.plugins: Dict[str, Type[Plugin]] = {}
        self.load_plugins()
    
    def load_plugins(self) -> None:
        sys.path.append(str(self.plugin_directory))
        
        for file in self.plugin_directory.glob("*.py"):
            if file.name.startswith("_"):
                continue
                
            module_name = file.stem
            module = importlib.import_module(module_name)
            
            for name, obj in inspect.getmembers(module):
                if (inspect.isclass(obj) and 
                    issubclass(obj, Plugin) and 
                    obj != Plugin):
                    self.plugins[name.lower()] = obj
    
    def create_plugin(self, plugin_name: str, *args, **kwargs) -> Plugin:
        plugin_class = self.plugins.get(plugin_name.lower())
        if not plugin_class:
            raise ValueError(f"Plugin {plugin_name} not found")
            
        plugin = plugin_class(*args, **kwargs)
        plugin.initialize()
        return plugin
    
    def list_available_plugins(self) -> List[str]:
        return list(self.plugins.keys())

# Example plugin implementation
class ImageProcessingPlugin(Plugin):
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.initialized = False
    
    def initialize(self) -> None:
        self.initialized = True
        print(f"Initialized {self.__class__.__name__}")
    
    def execute(self, image_path: str) -> str:
        if not self.initialized:
            raise RuntimeError("Plugin not initialized")
        
        # Simulate image processing
        return f"Processed {image_path} with {self.config}"
    
    def cleanup(self) -> None:
        self.initialized = False
        print(f"Cleaned up {self.__class__.__name__}")

# Example usage
plugins_dir = Path("./plugins")
factory = PluginFactory(plugins_dir)

# Create and use a plugin
config = {"resolution": "high", "format": "jpg"}
image_processor = factory.create_plugin("imageprocessing", config)
result = image_processor.execute("input.png")
image_processor.cleanup()

🚀 Plugin Factory Performance Analysis - Made Simple!

This slide shows you complete performance metrics and usage patterns for the Dynamic Plugin Factory System, including load times, memory usage, and execution performance across different plugin types.

Let’s break this down together! Here’s how we can tackle this:

import time
import psutil
import statistics
from typing import List, Dict

class PluginMetrics:
    def __init__(self, factory: PluginFactory):
        self.factory = factory
        self.metrics: Dict[str, List[float]] = {
            'load_time': [],
            'execution_time': [],
            'memory_usage': []
        }
    
    def measure_plugin_performance(self, plugin_name: str, iterations: int = 100) -> Dict[str, float]:
        process = psutil.Process()
        
        # Measure load time
        start_time = time.time()
        plugin = self.factory.create_plugin(plugin_name)
        load_time = time.time() - start_time
        self.metrics['load_time'].append(load_time)
        
        # Measure execution time and memory usage
        for _ in range(iterations):
            start_mem = process.memory_info().rss
            start_time = time.time()
            
            plugin.execute()
            
            exec_time = time.time() - start_time
            mem_used = (process.memory_info().rss - start_mem) / 1024 / 1024  # MB
            
            self.metrics['execution_time'].append(exec_time)
            self.metrics['memory_usage'].append(mem_used)
        
        return {
            'avg_load_time': statistics.mean(self.metrics['load_time']),
            'avg_execution_time': statistics.mean(self.metrics['execution_time']),
            'avg_memory_usage': statistics.mean(self.metrics['memory_usage']),
            'std_execution_time': statistics.stdev(self.metrics['execution_time'])
        }

# Example usage
factory = PluginFactory("./plugins")
metrics_analyzer = PluginMetrics(factory)

results = metrics_analyzer.measure_plugin_performance("imageprocessing")
print(f"""
Plugin Performance Metrics:
-------------------------
Average Load Time: {results['avg_load_time']:.3f} seconds
Average Execution Time: {results['avg_execution_time']:.3f} seconds
Average Memory Usage: {results['avg_memory_usage']:.2f} MB
Execution Time Std Dev: {results['std_execution_time']:.3f} seconds
""")

🚀 Extensible Report Generator Factory - Made Simple!

The Report Generator Factory shows you an cool implementation of the Factory pattern for creating different types of business reports, supporting multiple output formats and data sources while maintaining extensibility.

Here’s where it gets exciting! Here’s how we can tackle this:

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from dataclasses import dataclass
import json
import csv
import xml.etree.ElementTree as ET
from datetime import datetime

@dataclass
class ReportData:
    title: str
    data: List[Dict[str, Any]]
    timestamp: datetime
    metadata: Dict[str, Any]

class ReportGenerator(ABC):
    @abstractmethod
    def generate(self, data: ReportData) -> str:
        pass

class JSONReportGenerator(ReportGenerator):
    def generate(self, data: ReportData) -> str:
        report = {
            'title': data.title,
            'timestamp': data.timestamp.isoformat(),
            'metadata': data.metadata,
            'data': data.data
        }
        return json.dumps(report, indent=2)

class CSVReportGenerator(ReportGenerator):
    def generate(self, data: ReportData) -> str:
        output = []
        # Write headers
        if data.data:
            headers = list(data.data[0].keys())
            output.append(','.join(headers))
            
            # Write data rows
            for row in data.data:
                output.append(','.join(str(row[h]) for h in headers))
        
        return '\n'.join(output)

class XMLReportGenerator(ReportGenerator):
    def generate(self, data: ReportData) -> str:
        root = ET.Element('report')
        
        # Add metadata
        header = ET.SubElement(root, 'header')
        ET.SubElement(header, 'title').text = data.title
        ET.SubElement(header, 'timestamp').text = data.timestamp.isoformat()
        
        # Add data
        data_element = ET.SubElement(root, 'data')
        for item in data.data:
            record = ET.SubElement(data_element, 'record')
            for key, value in item.items():
                ET.SubElement(record, key).text = str(value)
        
        return ET.tostring(root, encoding='unicode', method='xml')

class ReportGeneratorFactory:
    def __init__(self):
        self._generators = {
            'json': JSONReportGenerator(),
            'csv': CSVReportGenerator(),
            'xml': XMLReportGenerator()
        }
    
    def register_generator(self, format_type: str, generator: ReportGenerator) -> None:
        self._generators[format_type.lower()] = generator
    
    def create_generator(self, format_type: str) -> ReportGenerator:
        generator = self._generators.get(format_type.lower())
        if not generator:
            raise ValueError(f"Unsupported format: {format_type}")
        return generator

# Example usage
sample_data = ReportData(
    title="Sales Report",
    data=[
        {"product": "Widget A", "sales": 100, "revenue": 1000},
        {"product": "Widget B", "sales": 150, "revenue": 1500}
    ],
    timestamp=datetime.now(),
    metadata={"author": "John Doe", "department": "Sales"}
)

factory = ReportGeneratorFactory()

# Generate reports in different formats
json_report = factory.create_generator('json').generate(sample_data)
csv_report = factory.create_generator('csv').generate(sample_data)
xml_report = factory.create_generator('xml').generate(sample_data)

print("JSON Report:", json_report)
print("\nCSV Report:", csv_report)
print("\nXML Report:", xml_report)

🚀 cool Configuration Factory - Made Simple!

The Configuration Factory pattern shows you a smart approach to managing application configurations across different environments, supporting multiple formats and dynamic updates.

This next part is really neat! Here’s how we can tackle this:

from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
import json
import yaml
import toml
from pathlib import Path
import threading
import time

class ConfigurationSource(ABC):
    @abstractmethod
    def load(self) -> Dict[str, Any]:
        pass
    
    @abstractmethod
    def save(self, config: Dict[str, Any]) -> None:
        pass

class JSONConfigurationSource(ConfigurationSource):
    def __init__(self, file_path: Path):
        self.file_path = file_path
    
    def load(self) -> Dict[str, Any]:
        with open(self.file_path) as f:
            return json.load(f)
    
    def save(self, config: Dict[str, Any]) -> None:
        with open(self.file_path, 'w') as f:
            json.dump(config, f, indent=2)

class YAMLConfigurationSource(ConfigurationSource):
    def __init__(self, file_path: Path):
        self.file_path = file_path
    
    def load(self) -> Dict[str, Any]:
        with open(self.file_path) as f:
            return yaml.safe_load(f)
    
    def save(self, config: Dict[str, Any]) -> None:
        with open(self.file_path, 'w') as f:
            yaml.dump(config, f)

class TOMLConfigurationSource(ConfigurationSource):
    def __init__(self, file_path: Path):
        self.file_path = file_path
    
    def load(self) -> Dict[str, Any]:
        with open(self.file_path) as f:
            return toml.load(f)
    
    def save(self, config: Dict[str, Any]) -> None:
        with open(self.file_path, 'w') as f:
            toml.dump(config, f)

class ConfigurationFactory:
    _instances: Dict[str, 'ConfigurationManager'] = {}
    _lock = threading.Lock()
    
    @classmethod
    def get_manager(cls, environment: str) -> 'ConfigurationManager':
        with cls._lock:
            if environment not in cls._instances:
                cls._instances[environment] = ConfigurationManager(environment)
            return cls._instances[environment]
    
    @staticmethod
    def create_source(file_path: Path) -> ConfigurationSource:
        extension = file_path.suffix.lower()
        if extension == '.json':
            return JSONConfigurationSource(file_path)
        elif extension in ('.yml', '.yaml'):
            return YAMLConfigurationSource(file_path)
        elif extension == '.toml':
            return TOMLConfigurationSource(file_path)
        raise ValueError(f"Unsupported configuration format: {extension}")

class ConfigurationManager:
    def __init__(self, environment: str):
        self.environment = environment
        self.sources: Dict[Path, ConfigurationSource] = {}
        self.config: Dict[str, Any] = {}
        self._lock = threading.Lock()
        self._watch_thread: Optional[threading.Thread] = None
        self._stop_watching = threading.Event()
    
    def add_source(self, file_path: Path) -> None:
        source = ConfigurationFactory.create_source(file_path)
        with self._lock:
            self.sources[file_path] = source
            self._update_config()
    
    def _update_config(self) -> None:
        new_config = {}
        for source in self.sources.values():
            new_config.update(source.load())
        self.config = new_config
    
    def get(self, key: str, default: Any = None) -> Any:
        return self.config.get(key, default)
    
    def start_watching(self, interval: float = 1.0) -> None:
        if self._watch_thread is not None:
            return
        
        def watch_sources():
            while not self._stop_watching.is_set():
                with self._lock:
                    self._update_config()
                time.sleep(interval)
        
        self._watch_thread = threading.Thread(target=watch_sources, daemon=True)
        self._watch_thread.start()
    
    def stop_watching(self) -> None:
        if self._watch_thread is not None:
            self._stop_watching.set()
            self._watch_thread.join()
            self._watch_thread = None

# Example usage
config_dir = Path("./config")
factory = ConfigurationFactory()

# Get development environment configuration manager
dev_config = factory.get_manager("development")
dev_config.add_source(config_dir / "config.dev.json")
dev_config.start_watching()

# Get production environment configuration manager
prod_config = factory.get_manager("production")
prod_config.add_source(config_dir / "config.prod.yaml")

print("Database URL (dev):", dev_config.get("database_url"))
print("Database URL (prod):", prod_config.get("database_url"))

🚀 Real-world Example - Content Management System Factory - Made Simple!

The CMS Factory pattern shows you a practical implementation for managing different types of content, including articles, videos, and podcasts, with support for versioning and content transformation.

Let me walk you through this step by step! Here’s how we can tackle this:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional, Any
import uuid

@dataclass
class ContentMetadata:
    author: str
    created_at: datetime
    tags: List[str]
    version: int
    status: str

class Content(ABC):
    def __init__(self, title: str, metadata: ContentMetadata):
        self.id = str(uuid.uuid4())
        self.title = title
        self.metadata = metadata
        self.versions: Dict[int, Any] = {}
    
    @abstractmethod
    def create_version(self) -> None:
        pass
    
    @abstractmethod
    def render(self) -> str:
        pass

class Article(Content):
    def __init__(self, title: str, body: str, metadata: ContentMetadata):
        super().__init__(title, metadata)
        self.body = body
        self.create_version()
    
    def create_version(self) -> None:
        self.versions[self.metadata.version] = {
            'title': self.title,
            'body': self.body,
            'metadata': self.metadata
        }
    
    def render(self) -> str:
        return f"""
        <article>
            <h1>{self.title}</h1>
            <div class="metadata">
                <span>Author: {self.metadata.author}</span>
                <span>Created: {self.metadata.created_at}</span>
                <span>Tags: {', '.join(self.metadata.tags)}</span>
            </div>
            <div class="content">{self.body}</div>
        </article>
        """

class Video(Content):
    def __init__(self, title: str, url: str, duration: int, metadata: ContentMetadata):
        super().__init__(title, metadata)
        self.url = url
        self.duration = duration
        self.create_version()
    
    def create_version(self) -> None:
        self.versions[self.metadata.version] = {
            'title': self.title,
            'url': self.url,
            'duration': self.duration,
            'metadata': self.metadata
        }
    
    def render(self) -> str:
        return f"""
        <video>
            <h2>{self.title}</h2>
            <div class="player" data-url="{self.url}">
                <span>Duration: {self.duration}s</span>
            </div>
            <div class="metadata">
                <span>Author: {self.metadata.author}</span>
                <span>Created: {self.metadata.created_at}</span>
            </div>
        </video>
        """

class ContentFactory:
    def __init__(self):
        self._content_types = {}
        self.register_default_types()
    
    def register_default_types(self):
        self._content_types['article'] = Article
        self._content_types['video'] = Video
    
    def register_content_type(self, type_name: str, content_class: type):
        self._content_types[type_name.lower()] = content_class
    
    def create_content(self, content_type: str, **kwargs) -> Content:
        content_class = self._content_types.get(content_type.lower())
        if not content_class:
            raise ValueError(f"Unsupported content type: {content_type}")
        
        metadata = ContentMetadata(
            author=kwargs.pop('author', 'Unknown'),
            created_at=kwargs.pop('created_at', datetime.now()),
            tags=kwargs.pop('tags', []),
            version=kwargs.pop('version', 1),
            status=kwargs.pop('status', 'draft')
        )
        
        return content_class(metadata=metadata, **kwargs)

# Example usage
factory = ContentFactory()

# Create an article
article = factory.create_content(
    'article',
    title="Understanding Factory Patterns",
    body="Factory patterns are essential...",
    author="John Doe",
    tags=['programming', 'design patterns']
)

# Create a video
video = factory.create_content(
    'video',
    title="Factory Pattern Tutorial",
    url="https://example.com/video.mp4",
    duration=300,
    author="Jane Smith",
    tags=['tutorial', 'programming']
)

print("Article HTML:")
print(article.render())
print("\nVideo HTML:")
print(video.render())

🚀 Data Processing Pipeline Factory - Made Simple!

This example showcases a flexible factory pattern for creating data processing pipelines, supporting various data sources and transformation steps with parallel processing capabilities.

Here’s where it gets exciting! Here’s how we can tackle this:

from abc import ABC, abstractmethod
from typing import List, Any, Callable
from concurrent.futures import ThreadPoolExecutor
import pandas as pd
import numpy as np
from dataclasses import dataclass
from queue import Queue
import threading

@dataclass
class ProcessingStep:
    name: str
    transform: Callable
    parallel: bool = False

class DataProcessor(ABC):
    @abstractmethod
    def process(self, data: Any) -> Any:
        pass

class CSVProcessor(DataProcessor):
    def __init__(self, steps: List[ProcessingStep]):
        self.steps = steps
        self.results_queue = Queue()
    
    def process(self, data: pd.DataFrame) -> pd.DataFrame:
        result = data.copy()
        
        with ThreadPoolExecutor() as executor:
            for step in self.steps:
                if step.parallel:
                    # Parallel processing for independent operations
                    futures = []
                    for column in result.columns:
                        future = executor.submit(
                            self._apply_transform,
                            result[column],
                            step.transform
                        )
                        futures.append((column, future))
                    
                    # Collect results
                    for column, future in futures:
                        result[column] = future.result()
                else:
                    # Sequential processing
                    result = step.transform(result)
                
                self.results_queue.put({
                    'step': step.name,
                    'shape': result.shape,
                    'dtypes': result.dtypes.to_dict()
                })
        
        return result

    @staticmethod
    def _apply_transform(data: pd.Series, transform: Callable) -> pd.Series:
        return transform(data)

class DataPipelineFactory:
    @staticmethod
    def create_pipeline(data_type: str, steps: List[ProcessingStep]) -> DataProcessor:
        if data_type.lower() == 'csv':
            return CSVProcessor(steps)
        raise ValueError(f"Unsupported data type: {data_type}")

# Example processing steps
def normalize_numeric(data: pd.DataFrame) -> pd.DataFrame:
    numeric_columns = data.select_dtypes(include=[np.number]).columns
    data[numeric_columns] = (data[numeric_columns] - data[numeric_columns].mean()) / data[numeric_columns].std()
    return data

def handle_missing_values(data: pd.DataFrame) -> pd.DataFrame:
    return data.fillna(data.mean())

def remove_outliers(series: pd.Series) -> pd.Series:
    if pd.api.types.is_numeric_dtype(series):
        z_scores = np.abs((series - series.mean()) / series.std())
        return series.mask(z_scores > 3, series.mean())
    return series

# Create processing pipeline
steps = [
    ProcessingStep("Missing Values", handle_missing_values),
    ProcessingStep("Normalization", normalize_numeric),
    ProcessingStep("Outlier Removal", remove_outliers, parallel=True)
]

# Example usage
factory = DataPipelineFactory()
pipeline = factory.create_pipeline('csv', steps)

# Sample data
data = pd.DataFrame({
    'A': [1, 2, None, 4, 100],
    'B': [5, 6, 7, None, 9],
    'C': [10, 11, 12, 13, 14]
})

# Process data
result = pipeline.process(data)

# Print processing results
while not pipeline.results_queue.empty():
    step_result = pipeline.results_queue.get()
    print(f"\nStep: {step_result['step']}")
    print(f"Output Shape: {step_result['shape']}")
    print("Output Types:")
    for col, dtype in step_result['dtypes'].items():
        print(f"  {col}: {dtype}")

🚀 Event Processing Factory System - Made Simple!

An cool implementation of the Factory pattern for handling different types of events in a distributed system, featuring event validation, transformation, and routing capabilities.

Ready for some cool stuff? Here’s how we can tackle this:

from abc import ABC, abstractmethod
from typing import Dict, Any, List, Callable, Optional
from datetime import datetime
import json
import hashlib
import threading
from queue import PriorityQueue
from dataclasses import dataclass
from enum import Enum, auto

class EventPriority(Enum):
    LOW = auto()
    MEDIUM = auto()
    HIGH = auto()
    CRITICAL = auto()

@dataclass
class EventMetadata:
    timestamp: datetime
    source: str
    priority: EventPriority
    correlation_id: str

class Event(ABC):
    def __init__(self, metadata: EventMetadata, payload: Dict[str, Any]):
        self.metadata = metadata
        self.payload = payload
        self.id = self._generate_id()
    
    @abstractmethod
    def validate(self) -> bool:
        pass
    
    @abstractmethod
    def transform(self) -> Dict[str, Any]:
        pass
    
    def _generate_id(self) -> str:
        content = f"{self.metadata.timestamp}{self.metadata.source}{str(self.payload)}"
        return hashlib.sha256(content.encode()).hexdigest()

class PaymentEvent(Event):
    def validate(self) -> bool:
        required_fields = {'amount', 'currency', 'payment_method'}
        return all(field in self.payload for field in required_fields)
    
    def transform(self) -> Dict[str, Any]:
        return {
            'event_type': 'payment',
            'amount_cents': int(float(self.payload['amount']) * 100),
            'currency': self.payload['currency'].upper(),
            'method': self.payload['payment_method'],
            'timestamp': self.metadata.timestamp.isoformat()
        }

class UserEvent(Event):
    def validate(self) -> bool:
        required_fields = {'user_id', 'action', 'details'}
        return all(field in self.payload for field in required_fields)
    
    def transform(self) -> Dict[str, Any]:
        return {
            'event_type': 'user_action',
            'user': self.payload['user_id'],
            'action_type': self.payload['action'],
            'details': json.dumps(self.payload['details']),
            'timestamp': self.metadata.timestamp.isoformat()
        }

class EventProcessor:
    def __init__(self):
        self.handlers: Dict[str, List[Callable]] = {}
        self.event_queue = PriorityQueue()
        self._running = False
        self._thread: Optional[threading.Thread] = None
    
    def register_handler(self, event_type: str, handler: Callable) -> None:
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    def process_event(self, event: Event) -> None:
        if not event.validate():
            raise ValueError(f"Invalid event: {event.id}")
        
        priority_value = {
            EventPriority.LOW: 4,
            EventPriority.MEDIUM: 3,
            EventPriority.HIGH: 2,
            EventPriority.CRITICAL: 1
        }[event.metadata.priority]
        
        self.event_queue.put((priority_value, event))
    
    def start_processing(self) -> None:
        if self._running:
            return
        
        self._running = True
        self._thread = threading.Thread(target=self._process_queue)
        self._thread.daemon = True
        self._thread.start()
    
    def stop_processing(self) -> None:
        self._running = False
        if self._thread:
            self._thread.join()
    
    def _process_queue(self) -> None:
        while self._running:
            try:
                _, event = self.event_queue.get(timeout=1)
                event_type = event.__class__.__name__
                
                transformed_data = event.transform()
                
                for handler in self.handlers.get(event_type, []):
                    try:
                        handler(transformed_data)
                    except Exception as e:
                        print(f"Error in handler {handler.__name__}: {e}")
                
                self.event_queue.task_done()
            except:
                continue

class EventFactory:
    _event_types: Dict[str, type] = {
        'payment': PaymentEvent,
        'user': UserEvent
    }
    
    @classmethod
    def register_event_type(cls, type_name: str, event_class: type) -> None:
        cls._event_types[type_name] = event_class
    
    @classmethod
    def create_event(cls, 
                    type_name: str, 
                    payload: Dict[str, Any],
                    source: str,
                    priority: EventPriority = EventPriority.MEDIUM) -> Event:
        if type_name not in cls._event_types:
            raise ValueError(f"Unknown event type: {type_name}")
        
        metadata = EventMetadata(
            timestamp=datetime.now(),
            source=source,
            priority=priority,
            correlation_id=hashlib.sha256(str(datetime.now().timestamp()).encode()).hexdigest()
        )
        
        return cls._event_types[type_name](metadata, payload)

# Example usage
def payment_handler(data: Dict[str, Any]) -> None:
    print(f"Processing payment: {data}")

def user_action_handler(data: Dict[str, Any]) -> None:
    print(f"Processing user action: {data}")

# Create event processor and register handlers
processor = EventProcessor()
processor.register_handler('PaymentEvent', payment_handler)
processor.register_handler('UserEvent', user_action_handler)

# Start event processing
processor.start_processing()

# Create and process events
factory = EventFactory()

payment_event = factory.create_event(
    'payment',
    {
        'amount': '99.99',
        'currency': 'usd',
        'payment_method': 'credit_card'
    },
    source='payment_gateway',
    priority=EventPriority.HIGH
)

user_event = factory.create_event(
    'user',
    {
        'user_id': '12345',
        'action': 'login',
        'details': {'ip': '192.168.1.1', 'device': 'mobile'}
    },
    source='auth_service',
    priority=EventPriority.LOW
)

processor.process_event(payment_event)
processor.process_event(user_event)

# Stop processing after events are handled
processor.stop_processing()

🚀 Additional Resources - Made Simple!

• Factory Pattern Research Papers and Articles:
• Design Patterns: Elements of Reusable Object-Oriented Software
  • Search: “Gang of Four Design Patterns Book”
• Modern Factory Patterns in Python: Best Practices and Implementation Strategies
  • Search: “Python Design Patterns - Factory Pattern Implementation”
• Implementing Abstract Factory Pattern in Distributed Systems
  • Search: “Distributed Systems Design Patterns”
• Online Resources:
• Python Design Patterns Documentation
  • https://python-patterns.guide/
• Real Python - Factory Pattern Tutorial
  • https://realpython.com/factory-pattern-python/
• Factory Pattern in Enterprise Applications
  • Search: “Enterprise Python Design Patterns”
• Community Resources:
• Python Design Patterns GitHub Repository
  • Search: “Python Design Patterns Examples”
• Stack Overflow Factory Pattern Questions
  • Search: “Python Factory Pattern Implementation Examples”
• Books and Publications:
• Python Design Patterns: For Sleek and Successful Development
  • Search: “Python Design Patterns Book”
• Design Patterns in Python: A Practical Guide with Examples
  • Search: “Practical Python Design Patterns”

🎊 Awesome Work!

You’ve just learned some really powerful techniques! Don’t worry if everything doesn’t click immediately - that’s totally normal. The best way to master these concepts is to practice with your own data.

What’s next? Try implementing these examples with your own datasets. Start small, experiment, and most importantly, have fun with it! Remember, every data science expert started exactly where you are right now.

Keep coding, keep learning, and keep being awesome! 🚀