From 2581be83613a76a61fd4707a2c110a2df6ab800d Mon Sep 17 00:00:00 2001 From: ProtossDragoon Date: Sat, 22 Jun 2024 16:23:54 +0900 Subject: [PATCH] =?UTF-8?q?Feat:=20=ED=94=BC=EC=B2=98=20=EC=8A=A4=ED=86=A0?= =?UTF-8?q?=EC=96=B4=20=EB=94=94=EB=A0=89=ED=86=A0=EB=A6=AC=EC=97=90?= =?UTF-8?q?=EC=84=9C=20=EA=B0=80=EC=9E=A5=20=EC=B5=9C=EC=8B=A0=EB=B2=84?= =?UTF-8?q?=EC=A0=84=20=EB=8B=A4=EC=9D=8C=20=EB=B2=84=EC=A0=84=20=EC=84=9C?= =?UTF-8?q?=EB=B8=8C=EB=94=94=EB=A0=89=ED=86=A0=EB=A6=AC=20=EA=B2=BD?= =?UTF-8?q?=EB=A1=9C=EB=A5=BC=20=EA=B0=80=EC=A0=B8=EC=98=A4=EB=8A=94=20?= =?UTF-8?q?=EA=B8=B0=EB=8A=A5=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- autosink_data_elt/path/autosink.py | 35 +++++++++++++++++++++++++++--- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git a/autosink_data_elt/path/autosink.py b/autosink_data_elt/path/autosink.py index 586a666..d574b91 100644 --- a/autosink_data_elt/path/autosink.py +++ b/autosink_data_elt/path/autosink.py @@ -71,16 +71,44 @@ def get_dataset_root_dir( ) @property - def latest_dataset_root_dir(self) -> Union[os.PathLike, str]: - """ Feature store 디렉토리의 서브디렉토리를 정렬했을 때 가장 latest 에 해당하는 서브디렉토리 경로 - """ + def available_versions_in_feature_store(self): available_versions = sorted(os.listdir(self.feature_store_dir)) if not available_versions: raise FileNotFoundError(f'경로 `{self.feature_store_dir}`에 디렉토리가 없습니다.') + return available_versions + + @property + def latest_dataset_root_dir(self) -> Union[os.PathLike, str]: + """ Feature store 디렉토리의 서브디렉토리를 정렬했을 때 가장 latest 에 해당하는 서브디렉토리 경로 + 예를 들어, 서브디렉토리에 v3, v4, v5 가 있다면 v5 의 경로를 반환한다. + """ + available_versions = self.available_versions_in_feature_store latest_version = available_versions[-1] logger.debug(f'사용 가능한 데이터셋 `{available_versions}`중 `{latest_version}`을 가져옵니다.') return self.get_dataset_root_dir(latest_version) + @property + def next_dataset_root_dir(self) -> Union[os.PathLike, str]: + """ Feature store 디렉토리의 서브디렉토리를 고려할 때 다음에 추가될 데이터셋 서브디렉토리 경로 + 예를 들어, 서브디렉토리에 v3, v4, v5 가 있다면 다음은 v6 이다. v6 의 경로를 반환한다. + 단, 이 함수를 실행한다고 해서 경로를 생성하지는 않는다. + NOTE: feature store 내 서브디렉토리는 항상 `v{n}` 의 형태로 이름붙여진다고 전제한다. + """ + available_versions = self.available_versions_in_feature_store + latest_version = available_versions[-1] + latest_version_number = int(latest_version[1:]) + next_version = f'v{latest_version_number + 1}' + ret = self.get_dataset_root_dir(next_version) + logger.debug( + f'사용 가능한 데이터셋이 `{available_versions}`이므로, ' + f'다음 데이터셋 버전 `{next_version}`의 경로를 반환합니다.' + ) + assert not os.path.exists(ret), ( + f'구현상 문제입니다. `{ret}` 경로가 생성되어서는 안됩니다. ' + '이때 경로가 생성되는 방식으로 구현된다면 로직이 꼬이게 됩니다.' + ) + return ret + if __name__ == '__main__': # path = AutosinkPath( @@ -105,3 +133,4 @@ def latest_dataset_root_dir(self) -> Union[os.PathLike, str]: print(path.feature_store_dir) print(path.get_dataset_root_dir('v3')) print(path.latest_dataset_root_dir) + print(path.next_dataset_root_dir)