| 39 | |
| 40 | |
| 41 | class Graph: |
| 42 | def __init__(self, graph_def, conn=None) -> None: |
| 43 | self._schema = GraphSchema() |
| 44 | self._schema.from_graph_def(graph_def) |
| 45 | self._conn: Connection = conn |
| 46 | self._schema._conn = conn |
| 47 | |
| 48 | def schema(self): |
| 49 | return self._schema |
| 50 | |
| 51 | def insert_vertex(self, vertex: VertexRecordKey, properties: dict): |
| 52 | return self.insert_vertices([[vertex, properties]]) |
| 53 | |
| 54 | def insert_vertices(self, vertices: list): |
| 55 | request = to_write_requests_pb("VERTEX", vertices, write_service_pb2.INSERT) |
| 56 | return self._conn.batch_write(request) |
| 57 | |
| 58 | def update_vertex_properties(self, vertex: VertexRecordKey, properties: dict): |
| 59 | request = to_write_requests_pb( |
| 60 | "VERTEX", [[vertex, properties]], write_service_pb2.UPDATE |
| 61 | ) |
| 62 | return self._conn.batch_write(request) |
| 63 | |
| 64 | def update_vertex_properties_batch(self, vertices: list): |
| 65 | request = to_write_requests_pb("VERTEX", vertices, write_service_pb2.UPDATE) |
| 66 | return self._conn.batch_write(request) |
| 67 | |
| 68 | def delete_vertex(self, vertex_pk: VertexRecordKey): |
| 69 | return self.delete_vertices([vertex_pk]) |
| 70 | |
| 71 | def delete_vertices(self, vertex_pks: list): |
| 72 | request = to_write_requests_pb( |
| 73 | "VERTEX", [[pk, {}] for pk in vertex_pks], write_service_pb2.DELETE |
| 74 | ) |
| 75 | return self._conn.batch_write(request) |
| 76 | |
| 77 | def insert_edge(self, edge: EdgeRecordKey, properties: dict): |
| 78 | return self.insert_edges([[edge, properties]]) |
| 79 | |
| 80 | def insert_edges(self, edges: list): |
| 81 | request = to_write_requests_pb("EDGE", edges, write_service_pb2.INSERT) |
| 82 | return self._conn.batch_write(request) |
| 83 | |
| 84 | def update_edge_properties(self, edge: EdgeRecordKey, properties: dict): |
| 85 | request = to_write_requests_pb( |
| 86 | "EDGE", [[edge, properties]], write_service_pb2.UPDATE |
| 87 | ) |
| 88 | return self._conn.batch_write(request) |
| 89 | |
| 90 | def update_edge_properties_batch(self, edges: list): |
| 91 | request = to_write_requests_pb("EDGE", edges, write_service_pb2.UPDATE) |
| 92 | return self._conn.batch_write(request) |
| 93 | |
| 94 | def delete_edge(self, edge: EdgeRecordKey): |
| 95 | return self.delete_edges([edge]) |
| 96 | |
| 97 | def delete_edges(self, edge_pks: list): |
| 98 | request = to_write_requests_pb( |
no outgoing calls
no test coverage detected