좋아요. 아래는 앞서 이야기한 has-a (조합형) 구조를 기반으로 한 완성된 설계 코드예요.Behavior 계층과 Storage 조립, 그리고 Memory/Pandas 백엔드 두 가지 예시를 모두 포함합니다.
🧩 구조 개요
storage/ ├── behaviors/ │ ├── base.py # 추상 Protocol 정의 │ ├── memory_impl.py # in-memory 구현 │ └── pandas_impl.py # pandas 기반 구현 ├── storage.py # Storage 조합 클래스 └── factory.py # backend 선택 팩토리
📘 1. behaviors/base.py
from typing import Protocol, Any, List
class SearchBehavior(Protocol):
def search(self, query: Any) -> List[Any]: ...
class AddBehavior(Protocol):
def add(self, item: Any) -> None: ...
class DeleteBehavior(Protocol):
def delete(self, key: Any) -> None: ...
class UpdateBehavior(Protocol):
def update(self, key: Any, value: Any) -> None: ...
class CountBehavior(Protocol):
def count(self) -> int: ...
📗 2. behaviors/memory_impl.py
from typing import Any, List
# --- In-Memory Behavior 구현 ---
class MemorySearch:
def __init__(self, store: list):
self.store = store
def search(self, predicate) -> List[Any]:
"""predicate는 lambda x: 조건 함수"""
return [item for item in self.store if predicate(item)]
class MemoryAdd:
def __init__(self, store: list):
self.store = store
def add(self, item: Any) -> None:
self.store.append(item)
class MemoryDelete:
def __init__(self, store: list):
self.store = store
def delete(self, key: Any) -> None:
self.store[:] = [i for i in self.store if i.get("id") != key]
class MemoryUpdate:
def __init__(self, store: list):
self.store = store
def update(self, key: Any, value: Any) -> None:
for item in self.store:
if item.get("id") == key:
item.update(value)
class MemoryCount:
def __init__(self, store: list):
self.store = store
def count(self) -> int:
return len(self.store)
📙 3. behaviors/pandas_impl.py
import pandas as pd
from typing import Any, List
# --- Pandas Behavior 구현 ---
class PandasSearch:
def __init__(self, df: pd.DataFrame):
self.df = df
def search(self, query: str) -> pd.DataFrame:
"""query는 pandas query 문법 문자열"""
return self.df.query(query).copy()
class PandasAdd:
def __init__(self, df: pd.DataFrame):
self.df = df
def add(self, item: dict) -> None:
self.df.loc[len(self.df)] = item
class PandasDelete:
def __init__(self, df: pd.DataFrame):
self.df = df
def delete(self, key: Any) -> None:
self.df.drop(self.df[self.df["id"] == key].index, inplace=True)
self.df.reset_index(drop=True, inplace=True)
class PandasUpdate:
def __init__(self, df: pd.DataFrame):
self.df = df
def update(self, key: Any, value: dict) -> None:
idx = self.df[self.df["id"] == key].index
for k, v in value.items():
self.df.loc[idx, k] = v
class PandasCount:
def __init__(self, df: pd.DataFrame):
self.df = df
def count(self) -> int:
return len(self.df)
📘 4. storage.py (조합형 Storage)
from typing import Optional
from .behaviors.base import (
SearchBehavior, AddBehavior, DeleteBehavior, UpdateBehavior, CountBehavior
)
class Storage:
"""조합형 Storage: 개별 Behavior 객체를 소유(has-a)"""
def __init__(
self,
*,
searcher: Optional[SearchBehavior] = None,
adder: Optional[AddBehavior] = None,
deleter: Optional[DeleteBehavior] = None,
updater: Optional[UpdateBehavior] = None,
counter: Optional[CountBehavior] = None,
):
self._searcher = searcher
self._adder = adder
self._deleter = deleter
self._updater = updater
self._counter = counter
# 단순 위임 메서드들
def search(self, query): return self._searcher.search(query)
def add(self, item): return self._adder.add(item)
def delete(self, key): return self._deleter.delete(key)
def update(self, key, value): return self._updater.update(key, value)
def count(self): return self._counter.count()
📕 5. factory.py (Storage 생성 팩토리)
import pandas as pd
from .storage import Storage
from .behaviors.memory_impl import (
MemorySearch, MemoryAdd, MemoryDelete, MemoryUpdate, MemoryCount,
)
from .behaviors.pandas_impl import (
PandasSearch, PandasAdd, PandasDelete, PandasUpdate, PandasCount,
)
class StorageFactory:
"""백엔드별 Storage 조립기"""
@staticmethod
def memory_backend() -> Storage:
data_store = []
return Storage(
searcher=MemorySearch(data_store),
adder=MemoryAdd(data_store),
deleter=MemoryDelete(data_store),
updater=MemoryUpdate(data_store),
counter=MemoryCount(data_store),
)
@staticmethod
def pandas_backend(columns=None) -> Storage:
df = pd.DataFrame(columns=columns or ["id", "name", "age"])
return Storage(
searcher=PandasSearch(df),
adder=PandasAdd(df),
deleter=PandasDelete(df),
updater=PandasUpdate(df),
counter=PandasCount(df),
)
🚀 6. 사용 예시
from storage.factory import StorageFactory
# --- In-memory backend ---
memory_storage = StorageFactory.memory_backend()
memory_storage.add({"id": 1, "name": "Alice", "age": 29})
memory_storage.add({"id": 2, "name": "Bob", "age": 32})
print(memory_storage.count()) # 2
print(memory_storage.search(lambda x: x["age"] > 30))
memory_storage.update(1, {"name": "Alicia"})
memory_storage.delete(2)
# --- Pandas backend ---
pandas_storage = StorageFactory.pandas_backend()
pandas_storage.add({"id": 1, "name": "Jane", "age": 24})
pandas_storage.add({"id": 2, "name": "Tom", "age": 35})
print(pandas_storage.search("age >= 30"))
print(pandas_storage.count()) # 2
✅ 구조적 특징 요약
| 항목 | 설명 |
|---|---|
| 조합형 설계 | Storage가 Behavior 객체를 “소유(has-a)” 하며 모든 동작을 위임 |
| 백엔드 분리 | Memory/Pandas 등 다양한 backend 교체 가능 |
| 테스트 용이성 | Behavior 단위로 독립 테스트 가능 |
| 확장성 | 새로운 backend(예: SQLite, Redis) 추가 시 factory만 확장 |
| 유연성 | 필요 시 Behavior 조합을 달리해 hybrid storage 구성 가능 |
원하신다면 이 설계를 기반으로
- ✅
FilterBehavior추가 (WHERE 절 파서) - ✅
TransactionBehavior(commit/rollback) - ✅
StorageAdapter(FastAPI나 CLI용 어댑터)
등의 확장 버전도 바로 이어서 설계해드릴 수 있어요.
원하시나요?
