複数のデータソースからファイルを取得する処理を抽象化して扱いたかったために書いてみました。
新しくデータソースが追加された際の登録し忘れを防止用にデコレータで登録できるようにしています。
すごくざっくりした解説ですがこんな感じで実装しています。
- データソースは
Storageクラスのサブクラスとして実装する - 新しいデータソースは
@StorageFactory.register()デコレータで登録する StorageFactory.create()でインスタンスを生成する
ソースコード
from abc import ABC, abstractmethod
from typing import Generator, Callable, MutableMapping, Mapping
class Storage(ABC):
def __init__(self, config: Mapping):
"""イニシャライザ
:param config: データ取得に必要な情報(URI, 認証情報...etc)
"""
self._config = config
@abstractmethod
def fetch(self) -> Generator[str, None, None]:
"""ダウンロードしたファイルのローカルパスを逐次返す"""
raise NotImplemented()
class StorageFactory:
# kind:Storageクラスの辞書
_storage: MutableMapping[str, type] = {}
@classmethod
def register(cls, kind: str) -> Callable:
"""Storageクラスをkindと紐づけて登録する"""
def wrapper(kls: type) -> type:
if not issubclass(kls, Storage):
raise Exception('Must be subclass of Storage class.')
cls._storage[kind] = kls
return kls
return wrapper
@classmethod
def create(cls, config: Mapping) -> Storage:
"""ファクトリメソッド"""
storage = cls._storage.get(config.get('kind'))
if storage:
return storage(config)
raise Exception('There is no storage.')
@StorageFactory.register('gcs')
class GCSStorage(Storage):
"""GCS用のStorageクラス"""
def fetch(self):
"""GCSからの取得処理"""
yield 'file_from_gcs'
@StorageFactory.register('s3')
class S3Storage(Storage):
"""S3用のStorageクラス"""
def fetch(self):
"""S3からの取得処理"""
yield 'file_from_s3'
def main():
gcs = StorageFactory.create({'kind': 'gcs', 'uri': 'gs://...'})
print(list(gcs.fetch()))
s3 = StorageFactory.create({'kind': 's3', 'uri': 's3://...'})
print(list(s3.fetch()))
if __name__ == '__main__':
main()
実行結果
こんな感じになります。
$ python3 factorymethod.py
['file_from_gcs']
['file_from_s3']
もっとこうしたほうがいいよーとかあればTwitterとかで@rhoboroにメンションしていただければ。