un level0

csst_dfs_commons/models/__init__.py

@@ -42,8 +42,10 @@ class Result(dict):
         return self
 
 class Request(object):
-    def __init__(self):
+    def __init__(self, **kvargs):
         super(Request, self).__init__()
+        for k, v in kvargs.items():
+            self.append(k, v)
 
     @staticmethod
     def from_dict(d: dict):
@@ -54,4 +56,12 @@ class Request(object):
 
     def append(self, k: str, v):
         self.__setattr__(k, v)
-        return self
\ No newline at end of file
+        return self
+
+    def __getattr__(self, attr):
+        try:
+            return object.__getattribute__(self, attr)
+        except:
+            return ''
+
+       
\ No newline at end of file

csst_dfs_commons/models/cpic.py

 import dataclasses
 from typing import Dict
 from .common import BaseModel, default_field
-
-@dataclasses.dataclass
-class Level0PrcRecord(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    pipeline_id: str = ""
-    prc_module: str = ""
-    params_file_path: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    result_file_path: str=""
-
-@dataclasses.dataclass
-class Level0Record(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    obs_id: str = ""
-    detector_no: str = ""
-    obs_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    detector_status_id: int = 0
-    filename: str=""
-    file_path: str=""
-    qc0_status: int = 0
-    qc0_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    create_time: str=""
-    header: Dict[str,object] = default_field({})
-
-@dataclasses.dataclass
-class CalMergeRecord(BaseModel):
-    id: int = 0
-    cal_id: str = ""
-    detector_no: str = ""
-    ref_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    filename: str=""
-    file_path: str=""
-    qc1_status: int = 0
-    qc1_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    prc_time: str=""
-    create_time: str=""
-    level0_ids: list = dataclasses.field(default_factory=list)
     
 @dataclasses.dataclass
 class Level1Record(BaseModel):

csst_dfs_commons/models/facility.py

 import dataclasses
-from typing import List
-from .common import BaseModel
+from typing import List,Dict
+from .common import BaseModel, default_field
 
 @dataclasses.dataclass
 class Observation(BaseModel):
@@ -90,3 +90,52 @@ class Level2ProducerRuning(BaseModel):
     prc_result: str=""
     create_time: str=""
     update_time: str=""        
+
+@dataclasses.dataclass
+class Level0PrcRecord(BaseModel):
+    id: int = 0
+    level0_id: str = ""
+    pipeline_id: str = ""
+    prc_module: str = ""
+    params_file_path: str=""
+    prc_status: int = 0
+    prc_time: str=""
+    result_file_path: str=""
+
+@dataclasses.dataclass
+class Level0Record(BaseModel):
+    id: int = 0
+    level0_id: str = ""
+    obs_id: str = ""
+    detector_no: str = ""
+    obs_type: str = ""
+    obs_time: str=""
+    exp_time: float = 0
+    detector_status_id: int = 0
+    filename: str=""
+    file_path: str=""
+    qc0_status: int = 0
+    qc0_time: str=""
+    prc_status: int = 0
+    prc_time: str=""
+    create_time: str=""
+    header: Dict[str,object] = default_field({})
+
+@dataclasses.dataclass
+class CalMergeRecord(BaseModel):
+    id: int = 0
+    cal_id: str = ""
+    detector_no: str = ""
+    ref_type: str = ""
+    obs_time: str=""
+    exp_time: float = 0
+    filename: str=""
+    file_path: str=""
+    qc1_status: int = 0
+    qc1_time: str=""
+    prc_status: int = 0
+    prc_time: str=""
+    prc_time: str=""
+    create_time: str=""
+    level0_ids: list = dataclasses.field(default_factory=list)
+    
\ No newline at end of file

csst_dfs_commons/models/ifs.py

@@ -2,53 +2,6 @@ import dataclasses
 from typing import Dict
 from .common import BaseModel, default_field
 
-@dataclasses.dataclass
-class Level0PrcRecord(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    pipeline_id: str = ""
-    prc_module: str = ""
-    params_file_path: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    result_file_path: str=""
-
-@dataclasses.dataclass
-class Level0Record(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    obs_id: str = ""
-    detector_no: str = ""
-    obs_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    detector_status_id: int = 0
-    filename: str=""
-    file_path: str=""
-    qc0_status: int = 0
-    qc0_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    create_time: str=""
-    header: Dict[str,object] = default_field({})
-
-@dataclasses.dataclass
-class CalMergeRecord(BaseModel):
-    id: int = 0
-    cal_id: str = ""
-    detector_no: str = ""
-    ref_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    filename: str=""
-    file_path: str=""
-    qc1_status: int = 0
-    qc1_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    prc_time: str=""
-    create_time: str=""
-    level0_ids: list = dataclasses.field(default_factory=list)
 @dataclasses.dataclass
 class Level1Record(BaseModel):
     id: int = 0

csst_dfs_commons/models/mci.py

@@ -2,53 +2,6 @@ import dataclasses
 from typing import Dict
 from .common import BaseModel, default_field
 
-@dataclasses.dataclass
-class Level0PrcRecord(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    pipeline_id: str = ""
-    prc_module: str = ""
-    params_file_path: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    result_file_path: str=""
-
-@dataclasses.dataclass
-class Level0Record(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    obs_id: str = ""
-    detector_no: str = ""
-    obs_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    detector_status_id: int = 0
-    filename: str=""
-    file_path: str=""
-    qc0_status: int = 0
-    qc0_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    create_time: str=""
-    header: Dict[str,object] = default_field({})
-
-@dataclasses.dataclass
-class CalMergeRecord(BaseModel):
-    id: int = 0
-    cal_id: str = ""
-    detector_no: str = ""
-    ref_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    filename: str=""
-    file_path: str=""
-    qc1_status: int = 0
-    qc1_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    prc_time: str=""
-    create_time: str=""
-    level0_ids: list = dataclasses.field(default_factory=list)
 @dataclasses.dataclass
 class Level1Record(BaseModel):
     id: int = 0

csst_dfs_commons/models/msc.py

@@ -2,54 +2,6 @@ import dataclasses
 from typing import Dict
 from .common import BaseModel, default_field
 
-@dataclasses.dataclass
-class Level0PrcRecord(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    pipeline_id: str = ""
-    prc_module: str = ""
-    params_file_path: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    result_file_path: str=""
-
-@dataclasses.dataclass
-class Level0Record(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    obs_id: str = ""
-    detector_no: str = ""
-    obs_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    detector_status_id: int = 0
-    filename: str=""
-    file_path: str=""
-    qc0_status: int = 0
-    qc0_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    create_time: str=""
-    header: Dict[str,object] = default_field({})
-
-@dataclasses.dataclass
-class CalMergeRecord(BaseModel):
-    id: int = 0
-    cal_id: str = ""
-    detector_no: str = ""
-    ref_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    filename: str=""
-    file_path: str=""
-    qc1_status: int = 0
-    qc1_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    prc_time: str=""
-    create_time: str=""
-    level0_ids: list = dataclasses.field(default_factory=list)
-    
 @dataclasses.dataclass
 class Level1Record(BaseModel):
     id: int = 0

csst_dfs_commons/models/sls.py

@@ -2,54 +2,6 @@ import dataclasses
 from typing import Dict
 from .common import BaseModel, default_field
 
-@dataclasses.dataclass
-class Level0PrcRecord(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    pipeline_id: str = ""
-    prc_module: str = ""
-    params_file_path: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    result_file_path: str=""
-
-@dataclasses.dataclass
-class Level0Record(BaseModel):
-    id: int = 0
-    level0_id: str = ""
-    obs_id: str = ""
-    detector_no: str = ""
-    obs_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    detector_status_id: int = 0
-    filename: str=""
-    file_path: str=""
-    qc0_status: int = 0
-    qc0_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    create_time: str=""
-    header: Dict[str,object] = default_field({})
-
-@dataclasses.dataclass
-class CalMergeRecord(BaseModel):
-    id: int = 0
-    cal_id: str = ""
-    detector_no: str = ""
-    ref_type: str = ""
-    obs_time: str=""
-    exp_time: float = 0
-    filename: str=""
-    file_path: str=""
-    qc1_status: int = 0
-    qc1_time: str=""
-    prc_status: int = 0
-    prc_time: str=""
-    prc_time: str=""
-    create_time: str=""
-    level0_ids: list = dataclasses.field(default_factory=list)
-
 @dataclasses.dataclass
 class Level1Record(BaseModel):
     id: int = 0

test/test_models.py

+import unittest
+from astropy.io import fits
+
+from csst_dfs_commons.models import Request
+
+class CommonModelsTestCase(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def test_request_result(self):
+        r = Request(id = 1)
+        assert r.id == 1, "not equal"
+        assert r.xxx == '', "not equal"