Loading csst_common/dfs.py +43 −5 Original line number Diff line number Diff line import os from csst_dfs_api.common.catalog import CatalogApi from csst_dfs_api.facility.level0 import Level0DataApi from csst_dfs_api.facility.level0prc import Level0PrcApi from csst_dfs_api.facility.level1 import Level1DataApi from csst_dfs_api.facility.level1prc import Level1PrcApi from csst_dfs_api.facility.otherdata import OtherDataApi from csst_dfs_api.mbi.level2 import Level2DataApi as MbiLevel2DataApi from csst_dfs_api.sls.level2spectra import Level2SpectraApi as SlsLevel2DataApi """ 1. DFS 配置 2. 接口retry功能 """ class DFS: def __init__(self, n_try=5): self.n_try = n_try pass @staticmethod def retry(func, n, *args, **kwargs): for attempt in range(1, n + 1): def retry(self, func, *args, **kwargs): for attempt in range(self.n_try): try: res = func(*args, **kwargs) assert res["code"] == 0, res Loading @@ -17,7 +27,35 @@ class DFS: print(f"Error occurs: {e.__repr__()}") raise RuntimeError("All attempts failed.") # DFS APIs @property def L0DataApi(self): return Level0DataApi() @property def L0PrcApi(self): return Level0PrcApi() @property def L1DataApi(self): return Level1DataApi() @property def L1PrcApi(self): return Level1PrcApi() @property def L2MbiDataApi(self): return MbiLevel2DataApi() @property def L2SlsDataApi(self): return SlsLevel2DataApi() @property def OtherDataApi(self): return OtherDataApi() @property def CatApi(self): return CatalogApi() Loading
csst_common/dfs.py +43 −5 Original line number Diff line number Diff line import os from csst_dfs_api.common.catalog import CatalogApi from csst_dfs_api.facility.level0 import Level0DataApi from csst_dfs_api.facility.level0prc import Level0PrcApi from csst_dfs_api.facility.level1 import Level1DataApi from csst_dfs_api.facility.level1prc import Level1PrcApi from csst_dfs_api.facility.otherdata import OtherDataApi from csst_dfs_api.mbi.level2 import Level2DataApi as MbiLevel2DataApi from csst_dfs_api.sls.level2spectra import Level2SpectraApi as SlsLevel2DataApi """ 1. DFS 配置 2. 接口retry功能 """ class DFS: def __init__(self, n_try=5): self.n_try = n_try pass @staticmethod def retry(func, n, *args, **kwargs): for attempt in range(1, n + 1): def retry(self, func, *args, **kwargs): for attempt in range(self.n_try): try: res = func(*args, **kwargs) assert res["code"] == 0, res Loading @@ -17,7 +27,35 @@ class DFS: print(f"Error occurs: {e.__repr__()}") raise RuntimeError("All attempts failed.") # DFS APIs @property def L0DataApi(self): return Level0DataApi() @property def L0PrcApi(self): return Level0PrcApi() @property def L1DataApi(self): return Level1DataApi() @property def L1PrcApi(self): return Level1PrcApi() @property def L2MbiDataApi(self): return MbiLevel2DataApi() @property def L2SlsDataApi(self): return SlsLevel2DataApi() @property def OtherDataApi(self): return OtherDataApi() @property def CatApi(self): return CatalogApi()