Commit 7dd5da6b authored by Wei Shoulin's avatar Wei Shoulin
Browse files

feat(level0): add batch update function for processing status

- Implement `update_prc_status_by_ids` function in level0.py for batch updating processing status of level 0 data by internal IDs
- Add corresponding unit test `test_update_prc_status_by_ids` in test_level0.py
parent d7ba5b77
Pipeline #9150 failed with stages
in 0 seconds
import os import os
from typing import Optional, Tuple, Literal from typing import Optional, Tuple, Literal, List
from .common import request, Result, utils, constants from .common import request, Result, utils, constants
DateTimeTuple = Tuple[str, str] DateTimeTuple = Tuple[str, str]
...@@ -140,6 +140,19 @@ def update_prc_status(level0_id: str, dag_run: str, prc_status: int, dataset: st ...@@ -140,6 +140,19 @@ def update_prc_status(level0_id: str, dag_run: str, prc_status: int, dataset: st
""" """
return request.put(f"/api/level0/prc_status/{level0_id}/{dag_run}", {'prc_status': prc_status, 'dataset': dataset}) return request.put(f"/api/level0/prc_status/{level0_id}/{dag_run}", {'prc_status': prc_status, 'dataset': dataset})
def update_prc_status_by_ids(ids: List[str], prc_status: int) -> Result:
"""
根据内部_id,批量更新0级数据的处理状态
Args:
ids (List[str]): 内部_id列表
prc_status (int): 处理状态
Returns:
Result: 操作结果
"""
return request.put("/api/level0/prc_status/batch", {'prc_status': prc_status, 'ids': ids})
def write(local_file: str, def write(local_file: str,
dataset: str = constants.DEFAULT_DATASET, dataset: str = constants.DEFAULT_DATASET,
**kwargs) -> Result: **kwargs) -> Result:
......
...@@ -39,6 +39,11 @@ class Level0TestCase(unittest.TestCase): ...@@ -39,6 +39,11 @@ class Level0TestCase(unittest.TestCase):
print(result) print(result)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message) self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
def test_update_prc_status_by_ids(self):
result = level0.update_prc_status_by_ids(ids = ["676ac74a530b47ca41568858"], prc_status=4)
print(result)
self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
def test_write(self): def test_write(self):
file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits" file_path = "/Users/wsl/temp/csst/import/CSST_MSC_MS_SCI_20240609181116_20240609181347_10109100157991_27_L0_V01.fits"
result = level0.write(local_file = file_path, dataset= 'msc-v093') result = level0.write(local_file = file_path, dataset= 'msc-v093')
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment