创建数据
(self,
data: Union[_TableType, List[_TableType]])
| 541 | |
| 542 | # 当仅传递一个数据时，返回插入数的 id或错误信息; 当传递多个数据时，返回插入的行数或错误信息 |
| 543 | def create(self, |
| 544 | data: Union[_TableType, List[_TableType]]) -> Union[int, str]: |
| 545 | """创建数据""" |
| 546 | if not isinstance(data, list): |
| 547 | data = [data] |
| 548 | |
| 549 | if not len(data): |
| 550 | raise ValueError("数据不能为空") |
| 551 | if not isinstance(data[0], self.data_cls): |
| 552 | raise ValueError("数据类型错误") |
| 553 | |
| 554 | now = int(time.time()) |
| 555 | |
| 556 | def fileter_data(item): |
| 557 | item_dict = item.to_dict() |
| 558 | if "id" in item_dict: |
| 559 | item_dict.pop("id") |
| 560 | if "created_at" in item_dict and item_dict["created_at"] is None: |
| 561 | item_dict["created_at"] = now |
| 562 | if "updated_at" in item_dict and item_dict["updated_at"] is None: |
| 563 | item_dict["updated_at"] = now |
| 564 | for fj_n in self.json_field_names: |
| 565 | if fj_n in item_dict and item_dict[fj_n] is not None and not isinstance(item_dict[fj_n], str): |
| 566 | item_dict[fj_n] = json.dumps(item_dict[fj_n]) |
| 567 | return item_dict |
| 568 | |
| 569 | data_list = list(map(fileter_data, data)) |
| 570 | if len(data_list) == 1: |
| 571 | try: |
| 572 | public.print_log(data_list) |
| 573 | res = self._db.table(self.table_name).insert(data_list[0]) |
| 574 | if isinstance(res, int): |
| 575 | return res |
| 576 | return str(res) |
| 577 | except Exception as e: |
| 578 | return str(e) |
| 579 | try: |
| 580 | res = self._db.table(self.table_name).batch_insert(data_list) |
| 581 | if isinstance(res, (int, bool)): |
| 582 | return len(data) |
| 583 | return str(res) |
| 584 | except Exception as e: |
| 585 | return str(e) |
| 586 | |
| 587 | def update(self, data: _TableType) -> str: |
| 588 | """更新数据""" |