(
record_attributes: list[str],
chunk_key: str,
)
| 44 | def is_float_regex(value): |
| 45 | return bool(re.match(r"^[-+]?[0-9]*\.?[0-9]+$", value)) |
| 46 | async def _handle_single_relationship_extraction( |
| 47 | record_attributes: list[str], |
| 48 | chunk_key: str, |
| 49 | ): |
| 50 | if len(record_attributes) < 5 or record_attributes[0] != '"relationship"': |
| 51 | return None |
| 52 | # add this record as edge |
| 53 | source = clean_str(record_attributes[1].upper()) |
| 54 | target = clean_str(record_attributes[2].upper()) |
| 55 | edge_description = clean_str(record_attributes[3]) |
| 56 | edge_source_id = chunk_key |
| 57 | weight = ( |
| 58 | float(record_attributes[-1]) if is_float_regex(record_attributes[-1]) else 1.0 |
| 59 | ) |
| 60 | return dict( |
| 61 | src_id=source, |
| 62 | tgt_id=target, |
| 63 | weight=weight, |
| 64 | description=edge_description, |
| 65 | source_id=edge_source_id, |
| 66 | ) |
no test coverage detected