🔤 第13课：Cypher查询

import re from typing import List, Dict, Any, Optional class CypherEngine: """简易Cypher查询引擎""" def __init__(self, graph): self.graph = graph def execute(self, query): ">>>执行Cypher查询（支持MATCH/WHERE/RETURN/ORDER BY/LIMIT）""" query = query.strip() if query.upper().startswith("MATCH"): return self._execute_match(query) elif query.upper().startswith("CREATE"): return self._execute_create(query) return {"error": "不支持的查询类型"} def _parse_match(self, query): """解析MATCH查询""" result = {"patterns": [], "where": None, "return_items": [], "order_by": None, "limit": None} # 提取MATCH部分 match_match = re.search(r'MATCH\s+(.+?)(?:\s+WHERE|\s+RETURN|$)', query, re.IGNORECASE) if match_match: pattern_str = match_match.group(1).strip() # 解析节点模式 (var:Label) node_pattern = re.findall(r'\((\w+)(?::(\w+))?\)', pattern_str) result["patterns"] = [(var, label) for var, label in node_pattern] # WHERE where_match = re.search(r'WHERE\s+(.+?)(?:\s+RETURN|$)', query, re.IGNORECASE) if where_match: result["where"] = where_match.group(1).strip() # RETURN return_match = re.search(r'RETURN\s+(.+?)(?:\s+ORDER|\s+LIMIT|$)', query, re.IGNORECASE) if return_match: items = return_match.group(1).strip().split(",") result["return_items"] = [item.strip() for item in items] # ORDER BY order_match = re.search(r'ORDER\s+BY\s+(.+?)(?:\s+LIMIT|$)', query, re.IGNORECASE) if order_match: result["order_by"] = order_match.group(1).strip() # LIMIT limit_match = re.search(r'LIMIT\s+(\d+)', query, re.IGNORECASE) if limit_match: result["limit"] = int(limit_match.group(1)) return result def _execute_match(self, query): parsed = self._parse_match(query) results = [] # 获取匹配的节点 bindings_list = [{}] for var, label in parsed["patterns"]: new_bindings = [] for bindings in bindings_list: nodes = self.graph.match_nodes(label) for nid, node in nodes: new_binding = dict(bindings) new_binding[var] = (nid, node) new_bindings.append(new_binding) bindings_list = new_bindings # WHERE过滤 if parsed["where"]: filtered = [] for bindings in bindings_list: try: where_str = parsed["where"] for var, (nid, node) in bindings.items(): for k, v in node["properties"].items(): where_str = where_str.replace(f"{var}.{k}", repr(v)) if eval(where_str): filtered.append(bindings) except: pass bindings_list = filtered # RETURN投影 for bindings in bindings_list: row = {} for item in parsed["return_items"]: if "." in item: var, prop = item.split(".", 1) if var in bindings: row[item] = bindings[var][1]["properties"].get(prop) elif item in bindings: row[item] = bindings[item][1]["properties"] results.append(row) # ORDER BY if parsed["order_by"]: desc = "DESC" in parsed["order_by"].upper() key = parsed["order_by"].replace("DESC", "").replace("ASC", "").strip() results.sort(key=lambda r: r.get(key, 0), reverse=desc) # LIMIT if parsed["limit"]: results = results[:parsed["limit"]] return results def _execute_create(self, query): """执行CREATE语句""" node_pattern = re.search(r'\((\w+):(\w+)\s*\{([^}]*)\}\)', query) if node_pattern: var, label, props_str = node_pattern.groups() properties = {} for pair in props_str.split(","): if ":" in pair: k, v = pair.split(":", 1) properties[k.strip()] = v.strip().strip(""'") nid = self.graph.create_node(label, properties) return {"created": nid, "properties": properties} return {"error": "无法解析CREATE语句"} # 使用前面定义的Neo4jStyleGraph class NG: """简化图用于演示Cypher""" def __init__(self): self.nodes = {} self._id = 0 self.label_index = {} def create_node(self, label, properties): nid = self._id; self._id += 1 self.nodes[nid] = {"labels": {label}, "properties": properties} self.label_index.setdefault(label, []).append(nid) return nid def match_nodes(self, label): return [(nid, self.nodes[nid]) for nid in self.label_index.get(label, [])] ng = NG() ng.create_node("Person", {"name": "鲁迅", "born": 1881}) ng.create_node("Person", {"name": "老舍", "born": 1899}) ng.create_node("Person", {"name": "徐志摩", "born": 1897}) ng.create_node("Book", {"name": "呐喊", "year": 1923}) ng.create_node("Book", {"name": "骆驼祥子", "year": 1937}) engine = CypherEngine(ng) # 测试查询 print("=== MATCH (n:Person) RETURN n.name, n.born ===") for row in engine.execute("MATCH (n:Person) RETURN n.name, n.born"): print(f" {row}") print(" === MATCH (n:Person) WHERE n.born > 1890 RETURN n.name ===") for row in engine.execute("MATCH (n:Person) WHERE n.born > 1890 RETURN n.name"): print(f" {row}") print(" === CREATE (n:Person {name: '巴金', born: 1904}) ===") print(engine.execute("CREATE (n:Person {name: '巴金', born: 1904})"))