문자열로 클래스 객체 동적 임포트 하여 사용하기

안녕하세요. 데이터 관리 업무를 맡고 있는 주니어 개발자 입니다. 최근 하나의 소스코드를 기반으로 다양한 브랜드가 수신 할 수 있는 프로세스를 구성 하고 있는데요. 첫 의도는 게이트웨이 패턴과 유사하게 흘러가게 기획을 했었는데, 점점 아키텍처가 난해해져 동적 임포트 방식을 이용하여 게이트웨이를 흉내내는 방식을 통해 객체 주입을 하였습니다. 하면서 처음 겪었던 부분들이 많아 내용을 남기고자 합니다.

필요한 객체들을 어떻게 보관할까?


제가 구성하는 프로세스에는 데이터를 매칭 시키고, 직접적으로 가공하는 부분이 존재합니다. 하나의 소스코드에서 성격이 다른 브랜드가 원하는 전문 양식대로 데이터를 가공 하려면 다형성을 이용할 수 있는데요. 그렇다고 한들, 이 오브젝트를 인스턴스화 해서 호출 할 땐 어떻게 해야할까요?

과연, 모든 인스턴스 데이터를 보관 하고 있어야 할지 아니면, 오브젝트 주소 값을 기준으로 디자인패턴 중 전략 패턴을 활용 할지 고민을 많이 했었는데요.

일단 제가 구상하는 내용은 이러했습니다.

기존 소스코드가 보관 되어 있는 방법을 보겠습니다.

class RealtimeConverterCallback:
    def __init__(self, args: Namespace, db: Database, schema: Dict[str, Union[Dict, str]]):
        self.args = args
        self.db = db
        self.converter = BrandOneConverter()

BrandOneConverter가 콜백 클래스가 초기화 될 때 생성되는 구조였습니다. 이 외에 다른 브랜드를 컨트롤 할 수 있는 컨버터 객체가 이 자리를 뚫고 들어와야했죠.

처음엔 DTO를 이용하여 프로퍼티로 가져올 수 있도록 해볼까 라는 고민도 했었는데요. 하지만, 이렇게 된다면 다른 브랜드가 생성 될 때 마다 객체를 DTO에 추가 하고, 브랜드명을 기준으로 해당 프로퍼티를 가져오는 코드도 필요했어요.

그래서, 파일 단위로 관리하며 추가로 생성 되더라도 소스코드는 변하지 않는 방법을 고민하다 선택한 것은 동적 임포트 입니다.

동적으로 객체 호출하기

class ConverterDependency:  

	def __init__(self, args: Namespace):  
        self.args = args  
        self.converter_name_prefix = "Converter"  
  
    def __call__(self, instance, master_data):  
        instance.converter = self.instance  
  
    @property  
    def module(self):  
        return self._import(self.args.brand)  
  
    @property  
    def class_object_name(self):  
        brand_name_captialize = self.args.brand.capitalize()  # NOTE: brand -> Brand
        return f"{brand_name_captialize}{self.converter_name_prefix}"  # NOTE: BrandConverter
  
    @property  
    def instance(self):  
        """  
        컨버터 인스턴스 가져오기  
        :return: 컨버터 인스턴스가 존재 하면 오브젝트 주소, 없다면 에러 반환  
        """
        if self.is_exists_converter():  
            return getattr(self.module, self.class_object_name)  
        else:  
            raise ValueError(f"{self.args.brand}의 컨버터 파일이 존재하지 않습니다")  
  
    @staticmethod  
    def _import(brand_name: str):  
        """  
        입력 받은 브랜드의 컨버트 파일 임포트  
        :param brand_name: brand name
        :return: module namespace  
        """  
        # NOTE: Convert하는 클래스 파일이 있는 디렉터리 위치  
        module_classes = Path(Path(src.__file__).parent, "converter", "classes")  
  
        # NOTE: {brand_name}으로 시작 하는 컨버트 파일 이름 추출 및 py 확장자 제거  
        module_file_name = "".join(list(filter(lambda x: x.startswith(brand_name), os.listdir(module_classes)))).strip(".py")  
  
        # NOTE: Import 형식 생성
        module_spec = str(Path(module_classes, module_file_name)).replace("/", ".")  
  
        # NOTE: import  
        return importlib.import_module(module_spec)  
  
    def is_exists_converter(self):  
        return hasattr(self.module, self.class_object_name)

제 흐름대로 만든 컨버터 의존성이라는 클래스를 생성 했는데요. 메서드 하나씩 보면서 흐름을 설명 해보겠습니다.

_init_

생성자에서 가장 최초로 호출 되어 인자로 넘겨지는 argparse.Namespace가 넘어오게 됩니다. 이 타입은 커맨드 라인으로 인자 값을 넘기고 저장된 Namedtuple과 유사한 형태입니다.

여기에 어느 브랜드의 수신 프로세스를 실행 할지 넘겨주며 콜백 객체에게 그대로 전달하면서 해당 브랜드만의 데이터 가공폼을 호출하려고 합니다.

@staticmethod: _import

실제 CPython이 실행 될 때 Parser에 의해 한 줄씩 문자열이 읽히게 되는걸로 기억하는데요. 그렇게 읽어들인 import line에 wraping 된 함수를 사용하기 위한 메서드라고 생각 하시면 되겠습니다.

객체를 호출 하기 전 알아두셔야 할 점은 브랜드별 데이터 가공 프로세스가 개별 파일로 관리 된다는점 입니다.

이렇게 3개의 브랜드가 있다면 폴더 구조는 이렇게 만들어집니다. 제가 필요한 정보는 A, B, C 가 되오니 이 필요한 값들을 문자열로 전달 받고 직접 모듈을 Import 하는 구조입니다.

흐름을 보자면 이러합니다.

  1. src/convert/classes 가 위치한 디렉터리 경로를 가져옵니다.
  2. 경로에서 원하는 브랜드 이름으로 시작하는 파일이 있는지 검색하고, .py를 제거합니다.
  3. 찾았던 모듈명과 전체 경로를 합친 후 디렉터리 구분자를 변경합니다. /-> .
  1. importlib 을 통해 문자열로 되어있는 모듈을 실제 모듈 스페이스가 호출 될 수 있도록 호출한 뒤 반환합니다.

@property: class_object_name

브랜드명과 프리픽스값을 결합 시킨 문자열을 반환 합니다. BrandConverter 가되도록 만들어주며 이 문자열은 실제 호출해야 할 클래스 객체 입니다.

@property: module

module 프로퍼티는 임포트한 모듈 네임스페이스 값이 들어있고 앞으로 오브젝트를 호출하기 위해 사용될 속성 값입니다. 여기서 반환 된 값은 실제 이 값이 들어있다고 보시면 됩니다. from src.convert import classes 에서 classes의 namespace 값을 반환한다고 보시면 되겠습니다. 이 값은 실제 모듈에서 클래스 객체를 가져올 때 아주 중요한 키 값으로 활용되는 것을 확인 할 수 있습니다.

is_exists_converter

속성 값으로 만들어 준 데이터 가공을 위한 클래스 객체가 Import한 모듈에 존재 하는지 검사를 거쳐 Boolean 값을 반환 합니다.

@property: instance

지금까지 모든 과정의 결과인 인스턴스화 된 데이터 컨버트 인스턴스를 반환하는 프로퍼티입니다. 만약 존재하지 않는다면 에러로 유인 하여 예외 처리를 해주면 되고, 실제로 객체가 있다면 인스턴스화 하여 콜백 메서드의 내부 속성 값에 넣어주게 됩니다.

동적 임포트 한 인스턴스 활용하기

class ConvertCallback:
      def __init__(self):
      self._converter = None
      self.injection_converter_dependency = ConverterDependency(args)  
      self.injection_converter_dependency(self, self.master_data)  

  @property  
  def converter(self):  
      return self._converter  

  @converter.setter  
  def converter(self, value):  
      self._converter = value
      
  # 데이터 가공

활용하는 곳에서 실제 데이터 가공 인스턴스를 setter로 넣어주면 해당 프로세스는 어느 브랜드가 들어와도 브랜드가 요구하는 전문 양식의 데이터 가공 클래스 파일만 생성 하면 어느 코드를 건들지 않아도 불러올 수 있게 됩니다.

테스트 케이스 수정 없이 통과 시키기

요약

  1. 문자열을 입력 받아 importlib 으로 모듈 네임스페이스를 가져옵니다.
  2. 모듈 네임스페이스에 원하는 클래스 객체가 있는지 검사합니다.
  1. 비즈니스 로직이 담긴 클래스에 게터-세터로 구성된 컨버터 속성 값에 인스턴스를 할당합니다.
  2. 할당된 인스턴스의 메인 로직을 실행합니다.

끝맺음

맨 처음 의도와는 달라지고, 이 코드도 구현하기 쉽게 최대한 수정이 되었는데요. 수정하기 전 마지막 의도는 클래스 데코레이터를 생성하여 사용하려고 했었습니다.

하지만, 객체간의 협력 구조를 갖고 있는 소스코드에서 데코레이터에 argparse.Namespace 타입을 넘겨줄 수 없었죠. 왜냐면 클래스 생성자 인자 값으로 아규먼트 네임스페이스를 전달 받아 사용해야 했었거든요.

그렇게 인자 값을 활용하기 까다로운 조건에서 쉽게 벗어나기 위해 게터 세터를 활용한 경우입니다.

감사합니다.