Skip to content

data_table

Native Layer v2 数据表模块

CDataTable <-> pd.DataFrame 转换，对外不暴露 C++ 对象。

NativeDataTableHandle

数据表不透明句柄，内部持有 C++ CDataTable 对象

Source code in dimine_python_sdk\lib\native\data_table.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
class NativeDataTableHandle:
    """数据表不透明句柄，内部持有 C++ CDataTable 对象"""

    def __init__(self, cpp_obj: Any):
        self._cpp_obj = cpp_obj

    @classmethod
    def create(cls) -> "NativeDataTableHandle":
        """创建空数据表"""
        return cls(Dm.CDataTable())

    @classmethod
    def from_dataframe(cls, df: "pd.DataFrame") -> "NativeDataTableHandle":
        """从 DataFrame 创建数据表"""
        require_pandas()
        import pandas as pd

        if not isinstance(df, pd.DataFrame):
            raise TypeError("df 必须为 pandas DataFrame")

        handle = cls.create()
        if len(df.columns) == 0:
            return handle

        for col in df.columns:
            handle.add_field(col, _infer_field_type_from_series(df[col]))

        for _, row in df.iterrows():
            record = handle.add_record()
            for col in df.columns:
                value = row[col]
                if pd.isna(value):
                    continue
                if isinstance(value, bool):
                    value = int(value)
                elif not isinstance(value, (str, int, float)):
                    value = str(value)
                record[col] = value

        return handle

    # ----------------------------------------------------------------------
    # 基本操作
    # ----------------------------------------------------------------------

    def load(self, file_path: str) -> None:
        """加载数据表文件"""
        result = self._cpp_obj.Load(file_path)
        if result is False:
            raise NativeDataTableError(f"加载数据表失败: {file_path}")

    def save(self, file_path: str | None = None) -> None:
        """保存数据表"""
        if file_path is not None:
            self._cpp_obj.SetFileName(file_path)
        self._cpp_obj.Save()
        #raise NativeDataTableError("保存数据表失败")

    def field_names(self) -> list[str]:
        """返回字段名列表"""
        count = self._cpp_obj.Get_Field_Count()
        return [self._cpp_obj.Get_Field_Name(i) for i in range(count)]

    def field_types(self) -> list[int]:
        """返回字段类型编码列表"""
        count = self._cpp_obj.Get_Field_Count()
        return [self._cpp_obj.Get_Field_Type(i) for i in range(count)]

    def field_definitions(self) -> list[NativeFieldDef]:
        """返回字段定义列表"""
        count = self._cpp_obj.Get_Field_Count()
        defs = []
        for i in range(count):
            name = self._cpp_obj.Get_Field_Name(i)
            ftype = self._cpp_obj.Get_Field_Type(i)
            defs.append(NativeFieldDef(name=name, type=ftype))
        return defs

    def record_count(self) -> int:
        """返回记录总数"""
        return self._cpp_obj.Get_Record_Count()

    def get_record(self, index: int) -> dict:
        """通过索引获取单条记录字典"""
        if index < 0 or index >= self.record_count():
            raise IndexError(f"记录索引 {index} 超出范围 [0, {self.record_count()})")
        return self._cpp_obj.Get_Record_Data(index)

    def add_field(self, name: str, field_type: int | str) -> None:
        """添加字段"""
        code = _resolve_field_type(field_type)
        self._cpp_obj.Add_Field(name, code)

    def add_record(self) -> "_RecordProxy":
        """添加一条新记录，返回记录代理"""
        record = self._cpp_obj.Add_Record()
        return _RecordProxy(record)

    def clear_records(self) -> None:
        """清空所有记录"""
        while self.record_count() > 0:
            self._cpp_obj.Del_Record(0)

    def to_dataframe(self) -> "pd.DataFrame":
        """转换为 pandas DataFrame"""
        require_pandas()
        import pandas as pd

        rows = [self.get_record(i) for i in range(self.record_count())]
        return pd.DataFrame(rows, columns=self.field_names())

    def insert_from_dataframe(
        self,
        df: "pd.DataFrame",
        column_mapping: dict[str, str] | None = None,
        exclude_columns: list[str] | None = None,
        field_types: dict[str, str] | None = None,
    ) -> None:
        """从 DataFrame 批量插入数据"""
        require_pandas()
        import pandas as pd

        column_mapping = column_mapping or {}
        exclude = set(exclude_columns or [])
        field_types = field_types or {}

        if self.record_count() == 0 and self._cpp_obj.Get_Field_Count() == 0:
            for col in df.columns:
                field_name = column_mapping.get(col, col)
                ftype = field_types.get(field_name)
                if ftype is None:
                    ftype = _infer_field_type_from_series(df[col])
                self.add_field(field_name, ftype)

        for _, row in df.iterrows():
            record = self.add_record()
            for col in df.columns:
                field_name = column_mapping.get(col, col)
                if field_name in exclude:
                    continue
                value = row[col]
                if pd.isna(value):
                    continue
                if isinstance(value, bool):
                    value = int(value)
                elif not isinstance(value, (str, int, float)):
                    value = str(value)
                record[field_name] = value

    def __len__(self) -> int:
        return self.record_count()

    def __iter__(self):
        for i in range(self.record_count()):
            yield self.get_record(i)

    def __getitem__(self, key: str | int | slice):
        if isinstance(key, str):
            return [row[key] for row in self]
        if isinstance(key, int):
            return self.get_record(key)
        if isinstance(key, slice):
            return [self.get_record(i) for i in range(*key.indices(len(self)))]
        raise TypeError(f"不支持的 key 类型: {type(key)}")

add_field(name, field_type)

添加字段

Source code in dimine_python_sdk\lib\native\data_table.py 
201
202
203
204
def add_field(self, name: str, field_type: int | str) -> None:
    """添加字段"""
    code = _resolve_field_type(field_type)
    self._cpp_obj.Add_Field(name, code)

add_record()

添加一条新记录，返回记录代理

Source code in dimine_python_sdk\lib\native\data_table.py 
206
207
208
209
def add_record(self) -> "_RecordProxy":
    """添加一条新记录，返回记录代理"""
    record = self._cpp_obj.Add_Record()
    return _RecordProxy(record)

clear_records()

清空所有记录

Source code in dimine_python_sdk\lib\native\data_table.py 
211
212
213
214
def clear_records(self) -> None:
    """清空所有记录"""
    while self.record_count() > 0:
        self._cpp_obj.Del_Record(0)

create() classmethod

创建空数据表

Source code in dimine_python_sdk\lib\native\data_table.py 
119
120
121
122
@classmethod
def create(cls) -> "NativeDataTableHandle":
    """创建空数据表"""
    return cls(Dm.CDataTable())

field_definitions()

返回字段定义列表

Source code in dimine_python_sdk\lib\native\data_table.py 
181
182
183
184
185
186
187
188
189
def field_definitions(self) -> list[NativeFieldDef]:
    """返回字段定义列表"""
    count = self._cpp_obj.Get_Field_Count()
    defs = []
    for i in range(count):
        name = self._cpp_obj.Get_Field_Name(i)
        ftype = self._cpp_obj.Get_Field_Type(i)
        defs.append(NativeFieldDef(name=name, type=ftype))
    return defs

field_names()

返回字段名列表

Source code in dimine_python_sdk\lib\native\data_table.py 
171
172
173
174
def field_names(self) -> list[str]:
    """返回字段名列表"""
    count = self._cpp_obj.Get_Field_Count()
    return [self._cpp_obj.Get_Field_Name(i) for i in range(count)]

field_types()

返回字段类型编码列表

Source code in dimine_python_sdk\lib\native\data_table.py 
176
177
178
179
def field_types(self) -> list[int]:
    """返回字段类型编码列表"""
    count = self._cpp_obj.Get_Field_Count()
    return [self._cpp_obj.Get_Field_Type(i) for i in range(count)]

from_dataframe(df) classmethod

从 DataFrame 创建数据表

Source code in dimine_python_sdk\lib\native\data_table.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@classmethod
def from_dataframe(cls, df: "pd.DataFrame") -> "NativeDataTableHandle":
    """从 DataFrame 创建数据表"""
    require_pandas()
    import pandas as pd

    if not isinstance(df, pd.DataFrame):
        raise TypeError("df 必须为 pandas DataFrame")

    handle = cls.create()
    if len(df.columns) == 0:
        return handle

    for col in df.columns:
        handle.add_field(col, _infer_field_type_from_series(df[col]))

    for _, row in df.iterrows():
        record = handle.add_record()
        for col in df.columns:
            value = row[col]
            if pd.isna(value):
                continue
            if isinstance(value, bool):
                value = int(value)
            elif not isinstance(value, (str, int, float)):
                value = str(value)
            record[col] = value

    return handle

get_record(index)

通过索引获取单条记录字典

Source code in dimine_python_sdk\lib\native\data_table.py 
195
196
197
198
199
def get_record(self, index: int) -> dict:
    """通过索引获取单条记录字典"""
    if index < 0 or index >= self.record_count():
        raise IndexError(f"记录索引 {index} 超出范围 [0, {self.record_count()})")
    return self._cpp_obj.Get_Record_Data(index)

insert_from_dataframe(df, column_mapping=None, exclude_columns=None, field_types=None)

从 DataFrame 批量插入数据

Source code in dimine_python_sdk\lib\native\data_table.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def insert_from_dataframe(
    self,
    df: "pd.DataFrame",
    column_mapping: dict[str, str] | None = None,
    exclude_columns: list[str] | None = None,
    field_types: dict[str, str] | None = None,
) -> None:
    """从 DataFrame 批量插入数据"""
    require_pandas()
    import pandas as pd

    column_mapping = column_mapping or {}
    exclude = set(exclude_columns or [])
    field_types = field_types or {}

    if self.record_count() == 0 and self._cpp_obj.Get_Field_Count() == 0:
        for col in df.columns:
            field_name = column_mapping.get(col, col)
            ftype = field_types.get(field_name)
            if ftype is None:
                ftype = _infer_field_type_from_series(df[col])
            self.add_field(field_name, ftype)

    for _, row in df.iterrows():
        record = self.add_record()
        for col in df.columns:
            field_name = column_mapping.get(col, col)
            if field_name in exclude:
                continue
            value = row[col]
            if pd.isna(value):
                continue
            if isinstance(value, bool):
                value = int(value)
            elif not isinstance(value, (str, int, float)):
                value = str(value)
            record[field_name] = value

load(file_path)

加载数据表文件

Source code in dimine_python_sdk\lib\native\data_table.py 
158
159
160
161
162
def load(self, file_path: str) -> None:
    """加载数据表文件"""
    result = self._cpp_obj.Load(file_path)
    if result is False:
        raise NativeDataTableError(f"加载数据表失败: {file_path}")

record_count()

返回记录总数

Source code in dimine_python_sdk\lib\native\data_table.py 
191
192
193
def record_count(self) -> int:
    """返回记录总数"""
    return self._cpp_obj.Get_Record_Count()

save(file_path=None)

保存数据表

Source code in dimine_python_sdk\lib\native\data_table.py 
164
165
166
167
168
def save(self, file_path: str | None = None) -> None:
    """保存数据表"""
    if file_path is not None:
        self._cpp_obj.SetFileName(file_path)
    self._cpp_obj.Save()

to_dataframe()

转换为 pandas DataFrame

Source code in dimine_python_sdk\lib\native\data_table.py 
216
217
218
219
220
221
222
def to_dataframe(self) -> "pd.DataFrame":
    """转换为 pandas DataFrame"""
    require_pandas()
    import pandas as pd

    rows = [self.get_record(i) for i in range(self.record_count())]
    return pd.DataFrame(rows, columns=self.field_names())

NativeFieldType

Bases: IntEnum

CDataTable 字段类型编码

Source code in dimine_python_sdk\lib\native\data_table.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class NativeFieldType(IntEnum):
    """CDataTable 字段类型编码"""

    UNKNOWN = 0
    BYTE = 1
    SHORT = 2
    INT = 3
    LONG = 4
    FLOAT = 5
    DOUBLE = 6
    STRING = 7
    COLOR = 8
    DATE = 9
    BINARY = 10
    LONG64 = 11