微信公众号: 人人都能python编程 ，欢迎分享，非公众号转载保留此声明。

设计模式其实是区分低等级程序员和高等级程序员的分水岭，当然这里的掌握程度不是只知道套用，这个是没用的，是要完全可以达到信手拈来的程度才行。除了基础知识和算法功力，设计模式则是软件开发中常见问题的可重用解决方案。通过使用这些模式，开发人员不必重复解决相同的问题，从而可以节省时间和精力。

这个就有点像武术里面的套路，一个“降龙十八掌“就可以走遍天下都不怕了。当然光知道套路也没用，需要实战，不然只有挨揍的份（想想马老师，会打闪电五连鞭的那个）！今天说说在Python FaskAPI里的一些套路，因为这个框架我用的比较多！先列出套路列表：

• 第一式：亢龙有悔 - 单例模式
• 第二式　飞龙在天 - 工厂模式
• 第三式　见龙在田 - 观察者模式
• 第四式　鸿渐于陆 - 策略模式

亢龙有悔 - 单例模式

单例模式是一种创建型设计模式，它确保一个类只有一个实例存在，并提供对其的全局访问点。当您需要确保在您的应用程序中只有一个类实例时（例如数据库连接或日志系统），此模式很有用。

实现一个单例模式其实非常简单，代码如下：

class Singleton:    
    _instance = None
    
    def __new__(cls):        
        if not cls._instance:            
            cls._instance = super().__new__(cls)        
        return cls._instance

您可以定义一个私有构造函数和一个创建并返回类实例的静态方法。静态方法应该检查实例是否已经创建，如果是，则返回它。如果不是，它应该创建一个新实例并返回它。该实例应存储为类中的私有静态变量。

我们来实战一下，在FastAPI里的情况：

from fastapi import FastAPI, Depends
from pymongo import MongoClient
from typing import Optional

class Database:    
    instance = None
    
    def __new__(cls):
        if cls.instance is None:            
            cls.instance = super().__new__(cls)
            cls.instance.client = MongoClient("mongodb://localhost:27017/")  
            cls.instance.db = cls.instance.client["mydatabase"]        
            return cls.instance
            
    def get_db() -> Optional[Database]:
        return Database()

app = FastAPI()
@app.get("/")
def read_root(db: Optional[Database] = Depends(get_db)):
    if db:        
        result = db.db.my_collection.find_one() 
        return {"message": result}
    else:        
        return {"message": "Failed to connect to database."}

稍微解释解释一下上面的代码，在此示例中，我们定义了一个数据库类，该类使用 PyMongo 库创建数据库连接。Database 类使用 Singleton 模式来确保在应用程序的生命周期内只创建该类的一个实例。

然后我们定义一个 get_db 函数，它返回数据库类的一个可选实例。通过将此函数作为参数与 Depends 函数一起传递，此函数用作我们的 FastAPI 路由处理程序中的依赖项。

调用 read_root 函数时，将解析 get_db 依赖项，如果 Database 类的实例不存在，则会创建它。然后该函数从数据库中的 my_collection 集合中检索单个文档并返回其内容。

通过以这种方式使用单例模式，我们可以确保在 FastAPI 应用程序的整个生命周期中只创建一个数据库连接，即使在使用依赖注入时也是如此。这有助于减少资源使用并提高性能，特别是对于需要频繁访问数据库的应用程序。

请注意，虽然此示例使用 PyMongo 库和 get_db 函数，但您可以将单例模式与 FastAPI 中的任何数据库库或依赖函数一起使用。

飞龙在天 - 工厂模式

工厂模式是一种创建型设计模式，它提供了一种创建对象的方法，而无需指定将要创建的对象的确切类。当您有一个类层次结构并且您希望根据某些输入（例如用户的选择或配置参数）创建不同类的对象时，此模式很有用。

class ShapeFactory:    
    @staticmethod    
    def create_shape(shape_type):
        if shape_type == "circle":
            return Circle()        
        elif shape_type == "square":
            return Square()        
        elif shape_type == "triangle":
            return Triangle()        
        else:            
            raise ValueError("Invalid shape type")

要实现工厂模式，您可以定义一个工厂方法，该方法根据某些输入创建并返回适当类的实例。该方法应该是静态的，并在单独的工厂类或类层次结构的基类中定义。

好了，简单的例子知道了，我们再来看在FastAPI里的情况：

from fastapi import FastAPI, HTTPException

class PaymentProcessor:
    def process_payment(self, amount: float):
        pass

class CreditCardProcessor(PaymentProcessor):
    def process_payment(self, amount: float):
        return f"Processing credit card payment for {amount} dollars."

class PayPalProcessor(PaymentProcessor):
    def process_payment(self, amount: float):
        return f"Processing PayPal payment for {amount} dollars."

class PaymentProcessorFactory:
    @staticmethod
    def create_processor(processor_type: str) -> PaymentProcessor:
        if processor_type == "credit_card":
            return CreditCardProcessor()
        elif processor_type == "paypal":
            return PayPalProcessor()
        else:
            raise HTTPException(status_code=400, detail="Invalid payment processor type.")

app = FastAPI()

@app.post("/process_payment/{processor_type}")
def process_payment(processor_type: str, amount: float):
    processor = PaymentProcessorFactory.create_processor(processor_type)
    return processor.process_payment(amount)

在此示例中，我们定义了一个表示支付处理器的 PaymentProcessor 抽象类，以及该类的两个具体实现：CreditCardProcessor 和 PayPalProcessor。然后我们定义一个 PaymentProcessorFactory 类，它使用工厂模式根据用户输入动态创建 CreditCardProcessor 和 PayPalProcessor 类的实例。

我们还定义了一个用于处理付款的路由处理程序，该处理程序采用 processor_type 参数和 amount 参数。process_payment 函数使用 PaymentProcessorFactory 根据 processor_type 参数创建相应支付处理器的实例，然后在创建的实例上调用 process_payment 方法来处理支付。

通过以这种方式使用工厂模式，我们可以根据用户输入动态创建对象，而无需将代码与对象的具体实现紧密耦合。随着时间的推移，这有助于使代码更加模块化并且更易于维护。

见龙在田 - 观察者模式

观察者模式是一种行为设计模式，它定义了对象之间的一对多依赖关系，这样当一个对象改变状态时，它的所有依赖对象都会收到通知并自动更新。当您有一组对象需要在某些其他对象更改时更新时，例如模型-视图-控制器体系结构，此模式很有用。先来一段简单的代码:

class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self):
        for observer in self._observers:
            observer.update()

class Observer:
    def update(self):
        pass

要实现观察者模式，您可以定义一个主题类，该类维护其依赖项列表并提供添加和删除依赖项的方法。主题类还应该提供一种方法来通知其依赖项发生变化。依赖者应该实现一个观察者接口，该接口定义了一种从主题接收更新的方法。

应该理解了吧，在FastAPI中实战一下

from fastapi import FastAPI
from typing import List

class Subject:
    def __init__(self):
        self.observers = []

    def register_observer(self, observer):
        self.observers.append(observer)

    def remove_observer(self, observer):
        self.observers.remove(observer)

    def notify_observers(self, message):
        for observer in self.observers:
            observer.update(message)

class Resource(Subject):
    def __init__(self):
        super().__init__()
        self.data = []

    def add_data(self, new_data):
        self.data.append(new_data)
        self.notify_observers(new_data)

class Subscriber:
    def __init__(self, name):
        self.name = name

    def update(self, message):
        print(f"{self.name} received message: {message}")

app = FastAPI()
resource = Resource()

@app.post("/add_data/{new_data}")
def add_data(new_data: str):
    resource.add_data(new_data)
    return {"message": "Data added."}

@app.post("/subscribe/{subscriber_name}")
def subscribe(subscriber_name: str):
    subscriber = Subscriber(subscriber_name)
    resource.register_observer(subscriber)
    return {"message": f"{subscriber_name} subscribed."}

@app.post("/unsubscribe/{subscriber_name}")
def unsubscribe(subscriber_name: str):
    subscriber = Subscriber(subscriber_name)
    resource.remove_observer(subscriber)
    return {"message": f"{subscriber_name} unsubscribed."}

在这个例子中，我们定义了一个 Subject 抽象类来表示一个可以观察的主题，以及一个 Resource 类来扩展 Subject 并表示一个可以修改的资源。我们还定义了一个表示资源观察者的订阅者类。

另外我们的 FastAPI 应用程序定义了三个路由：/add_data、/subscribe 和 /unsubscribe。当使用新数据字符串向 /add_data 端点发出 POST 请求时，我们 FastAPI 应用程序中的 add_data 函数将新数据添加到 Resource 并调用 notify_observers 方法来通知所有订阅者更改。

当使用订阅者名称字符串向 /subscribe 端点发出 POST 请求时，订阅函数会使用提供的名称创建一个新的订阅者对象，并将其注册为资源的观察者。

当使用订阅者名称字符串向 /unsubscribe 端点发出 POST 请求时，取消订阅函数会使用提供的名称创建一个新的订阅者对象，并将其作为资源的观察者删除。

通过以这种方式使用观察者模式，我们可以解耦主题和观察者对象，并允许多个订阅者接收有关资源更改的更新。随着时间的推移，这有助于提高代码的模块化和可维护性。

鸿渐于陆 - 策略模式

策略模式是一种行为设计模式，允许您定义一系列算法，封装每个算法，并使它们在运行时可互换。当您有一组执行类似任务但实现不同的算法并且您希望能够在它们之间轻松切换时，此模式很有用。

class Sort: 
    def sort(self, data): 
        pass

class QuickSort(Sort): 
    def sort(self, data): 
        # Implementation of quicksort algorithm pass

class MergeSort(Sort): 
    def sort(self, data): 
        # Implementation of mergesort algorithm pass

class Client: 
    def init(self, sort_strategy): 
        self.sort_strategy = sort_strategy

    def do_sort(self, data):
        self.sort_strategy.sort(data)

要实现策略模式，您可以定义一个接口或抽象类来定义执行任务的方法，然后定义一组实现该接口或扩展抽象类的具体类。客户端类应该维护对接口或抽象类实例的引用，并使用它来执行任务。

同样的，百读不如一练，放到FastAPI里看看这个套路怎么用！

from fastapi import FastAPI
from typing import List

class SortStrategy:
    def sort(self, data: List[int]) -> List[int]:
        pass

class BubbleSort(SortStrategy):
    def sort(self, data: List[int]) -> List[int]:
        n = len(data)
        for i in range(n):
            for j in range(n-i-1):
                if data[j] > data[j+1]:
                    data[j], data[j+1] = data[j+1], data[j]
        return data

class QuickSort(SortStrategy):
    def sort(self, data: List[int]) -> List[int]:
        if len(data) <= 1:
            return data
        pivot = data[len(data)//2]
        left = [x for x in data if x < pivot]
        middle = [x for x in data if x == pivot]
        right = [x for x in data if x > pivot]
        return self.sort(left) + middle + self.sort(right)

class Sorter:
    def __init__(self, strategy: SortStrategy):
        self.strategy = strategy

    def set_strategy(self, strategy: SortStrategy):
        self.strategy = strategy

    def sort(self, data: List[int]) -> List[int]:
        return self.strategy.sort(data)

app = FastAPI()
sorter = Sorter(BubbleSort())

@app.post("/sort/{strategy}")
def sort_data(strategy: str, data: List[int]):
    if strategy == "bubble":
        sorter.set_strategy(BubbleSort())
    elif strategy == "quick":
        sorter.set_strategy(QuickSort())
    else:
        return {"message": "Invalid sorting strategy."}
    sorted_data = sorter.sort(data)
    return {"sorted_data": sorted_data}

在此示例中，我们定义了代表分类策略的sorttrategy抽象类，以及类的两个具体实现：Bubblesort和QuickSort。然后，我们定义一个封装sorttrategy的分类器类，并提供了一种将分类为包封策略的排序方法。

我们的FastAPI应用程序定义了一个路由：/排序。当通过排序策略字符串和要排序的整数列表向 /排序端点提出帖子请求时，我们的fast_data函数在我们的fastapi应用程序中创建了一个新的分类器对象，并根据策略字符串参数使用适当的排序策略，然后调用魔力对象上的排序方法对数据进行排序。

通过以这种方式使用策略模式，我们可以封装不同的排序算法并使其可互换，从而使我们可以在不同的算法之间轻松切换不同的算法，而无需更改使用它们的代码。这可以有助于提高代码随时间的模块化和可维护性。

说在最后

正如文章开头说的，设计模式其实就是编程里的套路，套路就是要去练，在实践中和自己的其他技能进行结合。你甚至能创造一些套路，说不定下个套路的作者就是你的，期待你的反馈哦！