# pythonのinjectorを使用して、環境変数でインメモリデータベースへのアクセスとDynamoDBへのアクセスを切り替える
import abc
import os
import boto3
from injector import inject, Injector, Module, singleton
class Database(metaclass=abc.ABCMeta):
@abc.abstractmethod
def get(self, key):
pass
@abc.abstractmethod
def set(self, key, value):
pass
# テスト時のデータベース
class InMemoryDatabase(Database):
def __init__(self):
self.data = {}
def get(self, key):
return self.data.get(key)
def set(self, key, value):
self.data[key] = value
# 本番時のデータベース
class DynamoDB(Database):
def __init__(self):
session = boto3.Session(profile_name="")
self.client = session.client("dynamodb")
def get(self, key):
response = self.client.get_item(
TableName="dicontainer_test", Key={"pk": {"S": key}}
)
item = response.get("Item")
return item
def set(self, key, value):
self.client.put_item(
TableName="dicontainer_test",
Item={"key": {"S": key}, "value": {"S": value}},
)
# 依存の解決
class DataAccessModule(Module):
def configure(self, binder):
if os.environ.get("PRD"):
binder.bind(Database, to=DynamoDB, scope=singleton)
else:
binder.bind(Database, to=InMemoryDatabase, scope=singleton)
# DBを使ったビジネスロジックを記載するクラス
# DataAccessModule.configureで取得した環境変数によってコンストラクタに渡されるdatabaseインスタンスが変わる
class Service:
@inject
def __init__(self, database: Database):
self.database = database
def get_item(self):
return self.database.get("key1")
if __name__ == '__main__':
# 依存関係の登録
injector = Injector(modules=[DataAccessModule()])
# injector.getの引数に依存注入が必要なクラスを指定
service = injector.get(Service)
items = service.get_item()
print(items)
More than 1 year has passed since last update.
Register as a new user and use Qiita more conveniently
- You get articles that match your needs
- You can efficiently read back useful information
- You can use dark theme