Get an instance Protobuf RecordUpdate from an instance of Record (rec) Return a dictionary of necessary record items including the Protobuf RecordUpdate instance
(params, rec, **kwargs)
| 1022 | |
| 1023 | |
| 1024 | def get_pb2_record_update(params, rec, **kwargs): |
| 1025 | # type: (KeeperParams, Record, ...) -> Optional[dict] |
| 1026 | """Get an instance Protobuf RecordUpdate from an instance of Record (rec) |
| 1027 | |
| 1028 | Return a dictionary of necessary record items including the Protobuf RecordUpdate instance |
| 1029 | """ |
| 1030 | rec_data = prepare_record_v3(params, rec) |
| 1031 | if rec_data is None: |
| 1032 | return |
| 1033 | record_rq, audit = rec_data |
| 1034 | |
| 1035 | links_by_uid = kwargs.get('record_links_by_uid') |
| 1036 | if links_by_uid: |
| 1037 | links_add_by_uid = links_by_uid.get('record_links_add') or {} |
| 1038 | links_add = links_add_by_uid.get(rec.record_uid) or [] |
| 1039 | links_del_by_uid = links_by_uid.get('record_links_remove') or {} |
| 1040 | links_remove = links_del_by_uid.get(rec.record_uid) or [] |
| 1041 | else: |
| 1042 | links = kwargs.get('record_links') or {} |
| 1043 | links_add = links.get('record_links_add') or [] |
| 1044 | links_remove = links.get('record_links_remove') or [] |
| 1045 | |
| 1046 | record_links_add = [] |
| 1047 | for link in links_add: |
| 1048 | rl = record_pb2.RecordLink() |
| 1049 | rl.record_uid = link.get('record_uid') |
| 1050 | rl.record_key = link.get('record_key') |
| 1051 | if rl.record_uid and rl.record_key: |
| 1052 | record_links_add.append(rl) |
| 1053 | |
| 1054 | record_links_remove = [x['record_uid'] or b'' for x in links_remove] |
| 1055 | |
| 1056 | ru = record_pb2.RecordUpdate() |
| 1057 | uid = loginv3.CommonHelperMethods.url_safe_str_to_bytes(record_rq['record_uid']) |
| 1058 | data = loginv3.CommonHelperMethods.url_safe_str_to_bytes(record_rq['data']) |
| 1059 | ru.record_uid = uid |
| 1060 | ru.client_modified_time = record_rq['client_modified_time'] |
| 1061 | ru.revision = record_rq['revision'] |
| 1062 | ru.data = data |
| 1063 | #ru.non_shared_data = b'' |
| 1064 | if record_links_add: |
| 1065 | ru.record_links_add.extend(record_links_add) |
| 1066 | if record_links_remove: |
| 1067 | ru.record_links_remove.extend(record_links_remove) |
| 1068 | if audit: |
| 1069 | ru.audit.data = audit |
| 1070 | ru = attach_security_data(params, rec.record_uid, ru) |
| 1071 | |
| 1072 | record_rq['pb2_record_update'] = ru |
| 1073 | return record_rq |
| 1074 | |
| 1075 | |
| 1076 | def get_record_v3_response(params, rq, endpoint, record_rq_by_uid, silent=True): |
no test coverage detected