You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional, Union
|
|
|
|
from iti.common.enums import StorageTypeEnum
|
|
|
|
from .interface import StorageInterface
|
|
from .local import LocalStorage
|
|
|
|
|
|
class StorageManager:
|
|
_instances: dict[str, StorageInterface] = {}
|
|
|
|
@classmethod
|
|
def get_storage(
|
|
cls,
|
|
storage_type: Optional[Union[str, StorageTypeEnum]] = None,
|
|
*,
|
|
config: dict | None = None,
|
|
base_dir: str | os.PathLike | None = None,
|
|
) -> StorageInterface:
|
|
config = config or {}
|
|
storage_type_str = cls._normalize_storage_type(storage_type, config)
|
|
if storage_type_str not in cls._instances:
|
|
cls._instances[storage_type_str] = cls._create_storage(
|
|
storage_type_str,
|
|
config,
|
|
base_dir=base_dir,
|
|
)
|
|
return cls._instances[storage_type_str]
|
|
|
|
@staticmethod
|
|
def _normalize_storage_type(
|
|
storage_type: Optional[Union[str, StorageTypeEnum]],
|
|
config: dict,
|
|
) -> str:
|
|
if storage_type is None:
|
|
return config.get("DEFAULT_STORAGE_TYPE", StorageTypeEnum.LOCAL.value)
|
|
if isinstance(storage_type, StorageTypeEnum):
|
|
return storage_type.value
|
|
return storage_type
|
|
|
|
@staticmethod
|
|
def _create_storage(
|
|
storage_type: str,
|
|
config: dict,
|
|
*,
|
|
base_dir: str | os.PathLike | None = None,
|
|
) -> StorageInterface:
|
|
if storage_type == StorageTypeEnum.LOCAL.value:
|
|
local_config = dict(config.get("LOCAL", {}))
|
|
if not local_config.get("base_path"):
|
|
local_config["base_path"] = str(Path(base_dir or Path.cwd()) / "runtime" / "uploads")
|
|
return LocalStorage(local_config)
|
|
|
|
if storage_type == StorageTypeEnum.ALIYUN_OSS.value:
|
|
from .aliyun_oss import AliyunOSSStorage
|
|
|
|
return AliyunOSSStorage(config.get("ALIYUN_OSS", {}))
|
|
if storage_type == StorageTypeEnum.TENCENT_COS.value:
|
|
from .tencent_cos import TencentCOSStorage
|
|
|
|
return TencentCOSStorage(config.get("TENCENT_COS", {}))
|
|
if storage_type == StorageTypeEnum.QINIU_KODO.value:
|
|
from .qiniu_kodo import QiniuKodoStorage
|
|
|
|
return QiniuKodoStorage(config.get("QINIU_KODO", {}))
|
|
if storage_type == StorageTypeEnum.HUAWEI_OBS.value:
|
|
from .huawei_obs import HuaweiOBSStorage
|
|
|
|
return HuaweiOBSStorage(config.get("HUAWEI_OBS", {}))
|
|
if storage_type == StorageTypeEnum.MINIO.value:
|
|
from .minio_storage import MinIOStorage
|
|
|
|
return MinIOStorage(config.get("MINIO", {}))
|
|
if storage_type == StorageTypeEnum.AWS_S3.value:
|
|
raise NotImplementedError("AWS S3 适配器尚未实现")
|
|
raise ValueError(f"未支持的存储类型: {storage_type}")
|