메타데이터를 메모리에 할당하고 주입하기
안녕하세요. 저는 현재 데이터 관리 업무를 하고 있는 주니어 개발자입니다.
이번에 퓨어 파이썬으로 실시간 데이터 수신, 데이터 가공을 진행 할 때 데이터 수신부와 데이터 가공을 분리하고자 하는 요구사항을 적용 하면서 고민 해 보았던 내용입니다.
데이터를 가공 하기 위한 기초 메타데이터를 호출 하는 과정은 프로덕트 몽고 서버에 접근 하여 컬렉션에 있는 데이터와 실시간으로 수신 한 데이터의 고유 값이 존재 하는지 검사 해야 합니다.
실시간 데이터 수신량이 많기 때문에 매번 데이터베이스에 I/O를 하는 것은 옳지 않다고 생각이 되어 이런 방법으로 시도 해보고자 합니다.
실시간 데이터를 수신 한 이후 시점에서 가공 하기 위한 기초 메타데이터를 주입 하여 메모리에 할당 되어 있는 데이터와 실시간 데이터를 존재하는지 판단 하여 데이터 정보를 가져오는 것이 목적입니다.
클래스 속성 값 이용하기
class classproperty:
def __init__(self, method):
self.func = method
def __get__(self, instance, cls):
return self.func(cls)파이썬에서 @property 라는 인스턴스 변수를 설정 할 수 있는 데코레이터가 존재합니다.
클래스를 사용하기 위해 인스턴스화를 해야 하고 이후 메서드를 호출 해야 하는데 사실 이 클래스들은 그저 dataclass 와 별반 다를 것 없이 내부 속성 값에는 데이터만 존재 할 예정입니다.
그렇다면 적절히 클래스 변수를 이용 하여 데이터를 추가 한다면 클래스 자체만으로도 내부 속성 값에 접근 할 수 있게 됩니다.
클래스 속성 값 사용하기
class BaseCollection:
data = None
@classproperty
def has_preprocess_method(self) -> bool:
preprocess_method: Optional[classmethod] = self.__dict__.get("pre_process_pk_field")
return isinstance(preprocess_method, classmethod)
@classmethod
def pre_process_pk_field(cls, data):
pass
@classmethod
def pk_field_method(cls) -> Callable:
if cls.has_preprocess_method:
field: Callable = lambda x: cls.pre_process_pk_field(x)
else:
field: Callable = lambda x: x[cls.pk]
return field
class ChildCollection(BaseCollection):
collection_name = "test_collection"
pk = "_id"
@classmethod
def pre_process_pk_field(cls, data):
pk_field = str(data[cls.pk]).lstrip('0')
return pk_field 상속 시킬 베이스 클래스 하나를 선언 하여 내부에는 전처리가 필요한 컬렉션인지를 구분 합니다.이후 전처리가 필요한 컬렉션만 전처리를 진행 하게끔 만들기 위해 classproperty 에서 boolean 값을 반환 하는 것을 볼 수 있습니다.
그 반환 값을 사용 하는 곳은 pk_field_method 내부에서 조건 분기를 거친 후 lambda에 전처리 함수를 사용 할지, 기존 고유 값을 데이터로 사용할지 결정하게 됩니다.
번외
self.__dict__.get("pre_process_pk_field") 라는 코드를 사용 하기 전 geattr 을 이용하여 자식 클래스에서 전처리 함수가 있는지 판단을 시도 했었는데요.
getattr 을 사용 했을 때 부모 클래스에 속성 값을 온전히 자식 클래스에 전달 되었기 때문에 자식 클래스에서 전처리 함수가 없더라도 부모 클래스에 의해 빈 함수로 속성이 있다고 하며 의도하지 않은 결과 값을 반환 했습니다.
그 후, classproperty로 생성 된 속성 값에 의해 self.__dict__ 를 사용 하게 되면 인스턴스화 되지 않은 클래스에서 본인이 갖고 있는 classmethod 와 클래스 변수를 얻을 수 있었습니다.
만일, 인스턴스화 된 상태에서 self.__dict__ 를 사용한다면 인스턴스 변수가 존재 하기 때문에 인스턴스화 되지 않은 클래스 객체에서 사용 했었던 이유가 되었습니다.
이 부분은 의도에 맞게 수정하여 적용하면 자식 클래스에서 선언 한 메서드가 존재하는지 여부를 판단할 수 있었습니다.
__dict__가 아닌 __dir__() 을 사용 했다고 하더라도 의도에 맞게 작성 되었을 것이라고 생각이 듭니다.
이 때는 해당 인스턴스 내부에 사용 할 수 있는 모든 내용들이 list 타입이 반환 됩니다. in 을 사용 하여 메서드의 존재 여부를 조회가 가능하게 됩니다.
취향에 맞게 dir 을 통해 조건 분기로 존재하는 경우 Boolean Field를 반환 해도 되고, 저와 같이 __dict__ 를 사용하여 get 으로 접근 하여 결과 값을 반환해도 됩니다. 전 get에서 Key가 없는 경우 None으로 결과 값을 준다는 것을 이용 하여 isinstance 에서 타입 검사를 하면 필요 없는 조건을 줄일 수 있다고 생각 했기 때문에 딕셔너리 타입을 결정 하였습니다.
베이스 클래스 상속하기
class TestOneCollection(BaseCollection):
collection_name = "test_one_collection"
pk = "test_one_id"
class TestTwoCollection(BaseCollection):
collection_name = "test_two_collection"
pk = "test_two_id"
@classmethod
def pre_process_pk_field(cls, data):
pk_field = str(data[cls.pk]).lstrip('0')
return pk_field
class TestThreeCollection(BaseCollection):
collection_name = "test_three_collection"
pk = "test_three_id"
class TestFourCollection(BaseCollection):
collection_name = "test_four_collection"
query_condition = {"test_field": "test"}
pk = "test_four_id" 이렇게 베이스 클래스를 상속 받아 클래스 변수 값만 수정한다면 저희가 의도 했던 바 대로 전처리가 필요한 컬렉션, 필요하지 않은 컬렉션에 대해 유연하게 대응 할 수 있는 구조가 되었습니다.
이 소스코드 외에도 셋팅 설정이나 컬렉션에 따라 쿼리 조건을 변경 하여 베이스 클래스의 메서드 하나로 관리 할 수 있는 과정을 덧 붙혀서 코드의 이해를 돕도록 하겠습니다.
전체 코드로 내부 구조 이해하기
class BaseCollection:
data = None
collection_name: str
query_condition = {}
exclude = {"_id": 0}
pk: str
set_field = None
@classmethod
def set_context_data(cls):
collection_data: Dict[str, Any] = cls.query_set.select_master_data_find_all(cls.query_condition, cls.exclude)
result: Dict[str, Dict] = cls.set_collection_name_dict(collection_data)
setattr(cls, "data", result)
@classmethod
def set_collection_name_dict(cls, datas: Dict[str, Union[str, int]]) -> Dict[str, Dict]:
collection_name_dict: Dict = dict()
pk_field_method: Callable = cls.pk_field_method()
for data in datas:
pk_field = pk_field_method(data)
if cls.set_field:
collection_name_dict[pk_field] = data[cls.set_field]
else:
collection_name_dict[pk_field] = data
return collection_name_dict
@classmethod
def pre_process_pk_field(cls, data):
pass
@classmethod
def pk_field_method(cls) -> Callable:
is_preprocess_method_use: bool = cls.has_preprocess_method
if is_preprocess_method_use:
field: Callable = lambda x: cls.pre_process_pk_field(x)
else:
field: Callable = lambda x: x[cls.pk]
return field
@classproperty
def has_preprocess_method(self) -> bool:
print(self.__dict__)
preprocess_method: Optional[classmethod] = self.__dict__.get("pre_process_pk_field")
return isinstance(preprocess_method, classmethod)
@classproperty
def query_set(self) -> QueryManager: # note: 클래스 변수의 collection_name 을 기준으로 MongoCollection 생성
query = QueryManager
return query(self.collection_name)
@classmethod
def get_data(cls):
if cls.data is None:
cls.set_context_data()
return cls.data
class TestOneCollection(BaseCollection):
collection_name = "test_one_collection"
pk = "test_one_id"
class TestTwoCollection(BaseCollection):
collection_name = "test_two_collection"
pk = "test_two_id"
@classmethod
def pre_process_pk_field(cls, data):
pk_field = str(data[cls.pk]).lstrip('0')
return pk_field
class TestThreeCollection(BaseCollection):
collection_name = "test_three_collection"
pk = "test_three_id"
set_field = test_three_nm"
class TestFourCollection(BaseCollection):
collection_name = "test_four_collection"
query_condition = {"test_field": "test"}
pk = "test_four_id"
set_field = test_four_nm"설명을 덧 붙혀 추가 된 부분을 보시면 set_field, pk의 값을 기준으로 실제 메모리에 할당 하는 데이터 구조를 가공을 거친 후 저장 하는 것을 볼 수 있습니다.
이렇게 실시간 데이터를 쉽게 조회 하기 위해 가공 하는 과정의 반복을 최소화 하고 확장성을 고려하여 클래스를 설계한 구조 입니다. 그렇다면 이 데이터를 어떻게 사용하면 될까요?
마스터 데이터 생성
class MasterData(metaclass=Singleton):
master_data_classes = {
"test_one_collection": TestOneCollection,
"test_two_collection": TestTwoCollection,
"test_three_collection": TestThreeCollection,
"test_foure_collection": TestFourCollection,
}
def __str__(self):
docs = ", ".join([i for i in self.master_data_classes])
return f"{docs}"
def __call__(self, *args):
mst_cls = self.master_data_classes.get(*args)
if mst_cls:
mst_data = mst_cls.get_data()
setattr(self, *args, mst_data)
return mst_data
else:
raise ValueError(f"[Invalid Value] does not exists master table > {self.__str__()}")
if __name__ == '__main__':
master_data = MasterData()
print(master_data("test_one_collection"))
print(master_data("test_two_collection"))
print(master_data("test_three_collection"))
# print(master_data("query_mst")) # error!저는 마스터 데이터 객체를 모듈화 하여 실제 데이터를 가공 해야 하는 곳에 메타데이터 정보를 주입 하기 위해 객체를 생성 하였는데요. 하지만 이 때는 다른 클래스들과 달리 인스턴스화가 필요한 객체라는 것을 알 수 있습니다. 여기에 큰 의미는 없었습니다. 인스턴스화 된 클래스를 주입하는게 더 큰 관심사였을 뿐입니다.
위 소스코드에는 내용이 포함 되어 있지 않지만 실제 실무에서 적용 하기 위해 작성한 코드에는 기존 베이스 클래스의 메서드 중 set_collection_name_dict에서 반환된 데이터가 실제 필요한 데이터인데요.
이 데이터를 호출 할 때 get_item(self, collection_name, pk_value) 라는 메서드를 통해 호출을 하게 되고 이 메서드 내부는 실시간 데이터와 데이터베이스에 저장 되어 있는 메타데이터 정보를 비교 하는 로직과 데이터를 가져오는 로직이 들어가게 됩니다.
그렇다면 실제 사용 할 때는 master_data.get_item("test_one_collection", pk_value=1234567) 를 통해 실제 정보를 비교 하고 필요한 값을 가져올 수 있게 됩니다.
해당 클래스를 모듈로 사용하고, 호출되는 파이썬 코드에서 인스턴스화 된 마스터 데이터 클래스를 넘겨도 되고, 클래스만 넘겨도 되겠죠. 저는 클래스 객체를 파라미터로 넘겨서 호출 하는 코드에서 인스턴스화를 거친 후 사용 하는 방법을 사용 했습니다. 이러한 이유는 모듈에서 인스턴스화를 시키고 싶지 않았기 때문이었죠.
끝맺음
이렇게 파이썬 내부에서 데이터베이스 호출을 최소화 하며 메타데이터 정보를 통해 데이터를 가공 하는 경우일 때 메모리에 할당 한 후 주입 하는 과정을 적어봤는데요.
당연히 이 소스코드에 정답이 있는 것은 아니듯이 아직 제가 보는 시야가 매우 좁기 때문에 더 효율적인 코드는 아니라는 것 입니다.
그리고 또 많은 소스코드가 파이썬의 장고 프레임워크와 상당히 닮아 있는 것을 볼 수 있고, classproperty 또한 여기서 영감을 얻어 작성 하게 되었는데요.
오픈소스의 내부 구조를 읽고 실무에 적용 하다보니 장고는 되게 견고한 오픈소스라는 것을 알 수 있었던 계기가 되었습니다.
감사합니다.