Python-通过抽象基类实现存储抽象化

应用场景

场景：

张三需要将数据推送到rsync服务，于是通过rsync类进行推送
李四更喜欢ftp，于是又新增ftp类，修改rsync调用代码为ftp类进行推送
王五发现rsync和ftp出现了性能瓶颈，需要数据推送到S3、CEPH、COS等存储上面

问题：

代码耦合性太高，新增存储类型都需要修改多处代码
不能快速切换到各种存储类型
没有规范化方法名，导致不同人开发的类，方法名完全不一样

解决：

通过基类抽象化对象存储层
基类定义了通用方法，实现类必须实现基类定义的方法
通过类静态方法type，实现不同配置自动实例化到对应的类（策略模式）
新增类型，调用层新增相关类型即可，上层无感知

设计模式

定义基类

抽象基类，类似于java的interface，用于定义方法，而不实现方法。

from abc import ABC
from abc import abstractmethod


class ObjectStorage(ABC):
    @abstractmethod
    def upload(self, local_file_name: str, object_name: str) -> bool:
        pass

    @abstractmethod
    def download(self, object_name: str, local_file_name: str) -> bool:
        pass

    @abstractmethod
    def delete(self, object_name: str) -> bool:
        pass

    @abstractmethod
    def exist(self, object_name: str) -> bool:
        pass

    @staticmethod
    @abstractmethod
    def type() -> str:
        pass

RSYNC实现

实现参考：python调用rsync命令

class RSYNCStorage(ObjectStorage):
    def __init__(self, config: dict) -> None:
        pass

    def upload(self, local_file_name: str, object_name: str) -> bool:
        pass

    def download(self, object_name: str, local_file_name: str) -> bool:
        pass

    def delete(self, object_name: str) -> bool:
        pass

    def exist(self, object_name: str) -> bool:
        pass

    @staticmethod
    def type() -> str:
        return "rsync"

FTP实现

实现参考：https://docs.python.org/3/library/ftplib.html

class FTPStorage(ObjectStorage):
    def __init__(self, config: dict) -> None:
        pass

    def upload(self, local_file_name: str, object_name: str) -> bool:
        pass

    def download(self, object_name: str, local_file_name: str) -> bool:
        pass

    def delete(self, object_name: str) -> bool:
        pass

    def exist(self, object_name: str) -> bool:
        pass

    @staticmethod
    def type() -> str:
        return "ftp"

S3实现

实现参考：https://boto3.amazonaws.com/v1/documentation/api/latest/guide/s3-examples.html

class S3Storage(ObjectStorage):
    def __init__(self, config: dict) -> None:
        pass

    def upload(self, local_file_name: str, object_name: str) -> bool:
        pass

    def download(self, object_name: str, local_file_name: str) -> bool:
        pass

    def delete(self, object_name: str) -> bool:
        pass

    def exist(self, object_name: str) -> bool:
        pass

    @staticmethod
    def type() -> str:
        return "s3"

COS实现

实现参考：https://cloud.tencent.com/document/product/436/12269

class COSStorage(ObjectStorage):
    def __init__(self, config: dict) -> None:
        pass

    def upload(self, local_file_name: str, object_name: str) -> bool:
        pass

    def download(self, object_name: str, local_file_name: str) -> bool:
        pass

    def delete(self, object_name: str) -> bool:
        pass

    def exist(self, object_name: str) -> bool:
        pass

    @staticmethod
    def type() -> str:
        return "cos"

调用实现

策略模式：这里返回调用对象，因为都是ObjectStorage的实现类，所以可以调用ObjectStorage定义的方法。

def get_object_storage(config: dict) -> ObjectStorage:
    for s in (RSYNCStorage, FTPStorage, S3Storage, COSStorage):
        if s.type() == config.get("TYPE"):
            return s(config)
    raise Exception("Not found proper storage")

实例调用

初始化对象，并调用对象

cfg = {
    "TYPE": "rsync"
}
storage = get_object_storage(cfg)
print(storage.type())
# storage.upload()
# storage.download()
# storage.delete()
# storage.exist()

文章目录