Graph Databases: Modeling Complex Relationships

┌─────────────────────────────────────────────────────────────────────┐
│                    Graph Data Models                                 │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌──────────────────────┐    ┌──────────────────────┐              │
│  │   Property Graph     │    │   RDF Triple Store  │              │
│  ├──────────────────────┤    ├──────────────────────┤              │
│  │                      │    │                      │              │
│  │   (User)             │    │   Subject            │              │
│  │   - name             │    │     Predicate        │              │
│  │   - email            │────│     Object           │              │
│  │                      │    │                      │              │
│  │   (KNOWS)            │    │   (URI-based)       │              │
│  │   - since            │    │   (Semantic web)    │              │
│  │                      │    │                      │              │
│  └──────────────────────┘    └──────────────────────┘              │
│                                                                      │
│  Property Graph:                                                    │
│  - Nodes have properties                                            │
│  - Relationships have type and properties                           │
│  - Example: Neo4j, Amazon Neptune                                   │
│                                                                      │
│  RDF Triple Store:                                                  │
│  - Subject-Predicate-Object structure                              │
│  - URIs for global identification                                   │
│  - Example: Apache Jena, Blazegraph                                 │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

from dataclasses import dataclass
from typing import Dict, List, Optional, Any
from datetime import datetime

@dataclass
class Node:
    labels: List[str]
    properties: Dict[str, Any]
    id: Optional[str] = None
    
    def has_label(self, label: str) -> bool:
        return label in self.labels
    
    def get_property(self, key: str, default: Any = None) -> Any:
        return self.properties.get(key, default)

@dataclass
class Relationship:
    type: str
    start_node: Node
    end_node: Node
    properties: Dict[str, Any]
    id: Optional[str] = None

class PropertyGraph:
    def __init__(self):
        self.nodes: Dict[str, Node] = {}
        self.relationships: List[Relationship] = {}
        self.node_index: Dict[str, List[str]] = {}
        self.relationship_index: Dict[str, List[str]] = {}
    
    def add_node(self, node: Node) -> str:
        node_id = node.id or self._generate_id()
        node.id = node_id
        self.nodes[node_id] = node
        
        for label in node.labels:
            if label not in self.node_index:
                self.node_index[label] = []
            self.node_index[label].append(node_id)
        
        return node_id
    
    def add_relationship(self, rel: Relationship) -> str:
        rel_id = rel.id or self._generate_id()
        rel.id = rel_id
        
        self.relationships[rel_id] = rel
        
        rel_type = rel.type
        if rel_type not in self.relationship_index:
            self.relationship_index[rel_type] = []
        self.relationship_index[rel_type].append(rel_id)
        
        return rel_id
    
    def find_nodes(self, label: str, property_filter: Dict = None) -> List[Node]:
        node_ids = self.node_index.get(label, [])
        
        results = []
        for node_id in node_ids:
            node = self.nodes[node_id]
            if property_filter:
                if all(node.properties.get(k) == v for k, v in property_filter.items()):
                    results.append(node)
            else:
                results.append(node)
        
        return results
    
    def find_relationships(
        self, 
        start_node: Node = None, 
        end_node: Node = None,
        rel_type: str = None
    ) -> List[Relationship]:
        results = []
        
        for rel in self.relationships.values():
            if rel_type and rel.type != rel_type:
                continue
            if start_node and rel.start_node.id != start_node.id:
                continue
            if end_node and rel.end_node.id != end_node.id:
                continue
            results.append(rel)
        
        return results
    
    def _generate_id(self) -> str:
        import uuid
        return str(uuid.uuid4())

# Using Docker
docker run \
    --name neo4j \
    -p 7474:7474 -p 7687:7687 \
    -v neo4j/data:/data \
    -v neo4j/logs:/logs \
    -e NEO4J_AUTH=neo4j/password \
    neo4j:latest

# Using Neo4j Desktop (recommended for development)
# Download from https://neo4j.com/download/

from neo4j import GraphDatabase

class Neo4jConnection:
    def __init__(self, uri: str, user: str, password: str):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def close(self):
        self.driver.close()
    
    def execute(self, query: str, parameters: Dict = None):
        with self.driver.session() as session:
            result = session.run(query, parameters or {})
            return [record.data() for record in result]
    
    def execute_single(self, query: str, parameters: Dict = None):
        with self.driver.session() as session:
            result = session.run(query, parameters or {})
            return result.single()

# Example queries
queries = {
    "create_node": """
        CREATE (p:Person {name: $name, email: $email})
        RETURN p
    """,
    
    "create_relationship": """
        MATCH (a:Person {name: $person1})
        MATCH (b:Person {name: $person2})
        CREATE (a)-[r:KNOWS {since: $since}]->(b)
        RETURN r
    """,
    
    "find_friends": """
        MATCH (person:Person {name: $name})-[:KNOWS]->(friend)
        RETURN friend.name AS friendName, friend.email AS friendEmail
    """,
    
    "find_friends_of_friends": """
        MATCH (person:Person {name: $name})-[:KNOWS]->()-[:KNOWS]->(fof)
        WHERE NOT (person)-[:KNOWS]->(fof)
        AND person <> fof
        RETURN DISTINCT fof.name AS name, count(*) AS commonFriends
        ORDER BY commonFriends DESC
    """,
    
    "recommend_friends": """
        MATCH (person:Person {name: $name})-[:KNOWS]->(friend)
        MATCH (friend)-[:KNOWS]->(suggestion)
        WHERE NOT (person)-[:KNOWS]->(suggestion)
        AND person <> suggestion
        RETURN suggestion.name AS suggestedFriend, 
               count(*) AS mutualFriends
        ORDER BY mutualFriends DESC
        LIMIT 5
    """,
    
    "shortest_path": """
        MATCH (start:Person {name: $startName}),
              (end:Person {name: $endName})
        MATCH path = shortestPath((start)-[*]-(end))
        RETURN path, length(path) AS distance
    """,
    
    "graph_algorithms": """
        CALL gds.graph.project(
            'myGraph',
            'Person',
            'KNOWS',
            {relationshipProperties: 'weight'}
        )
        CALL gds.pageRank.write('myGraph', {writeProperty: 'pageRank'})
        YIELD nodePropertiesWritten, ranIterations
        RETURN nodePropertiesWritten, ranIterations
    """
}

class SocialNetworkGraph:
    def __init__(self, neo4j: Neo4jConnection):
        self.db = neo4j
    
    def create_user(self, name: str, email: str, bio: str = None) -> Dict:
        query = """
            CREATE (u:User {
                name: $name, 
                email: $email, 
                bio: $bio,
                createdAt: datetime()
            })
            RETURN u
        """
        result = self.db.execute_single(query, {"name": name, "email": email, "bio": bio})
        return result["u"] if result else None
    
    def add_friend(self, user1: str, user2: str) -> bool:
        query = """
            MATCH (u1:User {name: $user1})
            MATCH (u2:User {name: $user2})
            WHERE NOT (u1)-[:KNOWS]->(u2)
            CREATE (u1)-[:KNOWS {since: datetime()}]->(u2)
            RETURN true AS created
        """
        result = self.db.execute_single(query, {"user1": user1, "user2": user2})
        return result and result.get("created", False)
    
    def get_mutual_friends(self, user1: str, user2: str) -> List[str]:
        query = """
            MATCH (u1:User {name: $user1})-[:KNOWS]->(friend)<-[:KNOWS]-(u2:User {name: $user2})
            RETURN friend.name AS name
        """
        return [r["name"] for r in self.db.execute(query, {"user1": user1, "user2": user2})]
    
    def suggest_friends(self, user: str, limit: int = 10) -> List[Dict]:
        query = """
            MATCH (user:User {name: $user})-[:KNOWS]->(currentFriend)
            MATCH (currentFriend)-[:KNOWS]->(suggestion)
            WHERE NOT (user)-[:KNOWS]->(suggestion)
            AND user <> suggestion
            WITH suggestion, count(currentFriend) AS mutualFriends
            RETURN suggestion.name AS name, suggestion.email AS email, mutualFriends
            ORDER BY mutualFriends DESC
            LIMIT $limit
        """
        return self.db.execute(query, {"user": user, "limit": limit})
    
    def get_user_influence(self, user: str) -> Dict:
        query = """
            MATCH (user:User {name: $user})
            OPTIONAL MATCH (user)-[:KNOWS]->(direct)<-[:KNOWS]-(indirect)
            WITH user, count(DISTINCT direct) AS directCount, count(DISTINCT indirect) AS indirectCount
            RETURN directCount + indirectCount AS totalReach,
                   directCount AS directFriends,
                   indirectCount AS indirectFriends
        """
        result = self.db.execute_single(query, {"user": user})
        return result if result else {}

class KnowledgeGraph:
    def __init__(self, neo4j: Neo4jConnection):
        self.db = neo4j
    
    def add_entity(self, entity_type: str, name: str, properties: Dict) -> Dict:
        query = f"""
            CREATE (e:{entity_type} {{name: $name}})
            SET e += $properties
            RETURN e
        """
        return self.db.execute_single(query, {"name": name, "properties": properties})
    
    def add_relationship(
        self, 
        from_entity: str, 
        from_type: str, 
        relationship: str, 
        to_entity: str, 
        to_type: str,
        properties: Dict = None
    ) -> bool:
        query = f"""
            MATCH (from:{from_type} {{name: $fromEntity}})
            MATCH (to:{to_type} {{name: $toEntity}})
            CREATE (from)-[r:{relationship} $props]->(to)
            RETURN r IS NOT NULL AS created
        """
        return self.db.execute_single(
            query, 
            {"fromEntity": from_entity, "toEntity": to_entity, "props": properties or {}}
        )
    
    def find_related_entities(self, entity: str, relationship: str, depth: int = 1) -> List[Dict]:
        query = f"""
            MATCH (start {{name: $entity}})-[r:{relationship}*1..{depth}]->(related)
            RETURN DISTINCT related.name AS name, 
                   labels(related)[0] AS type,
                   length(r) AS distance
            ORDER BY distance, name
        """
        return self.db.execute(query, {"entity": entity})
    
    def find_path(self, from_entity: str, to_entity: str) -> List[Dict]:
        query = """
            MATCH (start {name: $fromEntity}), (end {name: $toEntity})
            MATCH path = allShortestPaths((start)-[*]-(end))
            RETURN path, length(path) AS distance
            LIMIT 1
        """
        return self.db.execute(query, {"fromEntity": from_entity, "toEntity": to_entity})
    
    def get_entity_centrality(self) -> List[Dict]:
        query = """
            CALL gds.degree.stream('myGraph')
            YIELD nodeId, score
            WITH gds.util.asNode(nodeId) AS node, score
            RETURN node.name AS entity, score AS connections
            ORDER BY connections DESC
            LIMIT 20
        """
        return self.db.execute(query)

class GraphModeler:
    @staticmethod
    def model_from_requirements(requirements: Dict) -> Dict:
        entities = requirements.get("entities", [])
        relationships = requirements.get("relationships", [])
        
        nodes = []
        for entity in entities:
            node = {
                "label": entity["name"],
                "description": entity.get("description", ""),
                "properties": [
                    {"name": "id", "type": "STRING", "unique": True},
                ],
                "identifying_property": "id"
            }
            
            for attr in entity.get("attributes", []):
                node["properties"].append({
                    "name": attr["name"],
                    "type": attr.get("type", "STRING"),
                    "required": attr.get("required", False)
                })
            
            nodes.append(node)
        
        rels = []
        for relationship in relationships:
            rel = {
                "type": relationship["name"].upper(),
                "from": relationship["from"],
                "to": relationship["to"],
                "description": relationship.get("description", ""),
                "properties": []
            }
            
            for prop in relationship.get("properties", []):
                rel["properties"].append({
                    "name": prop["name"],
                    "type": prop.get("type", "STRING")
                })
            
            rels.append(rel)
        
        return {"nodes": nodes, "relationships": rels}
    
    @staticmethod
    def generate_cypher_schema(model: Dict) -> str:
        cypher = ""
        
        for node in model["nodes"]:
            props = []
            for prop in node["properties"]:
                prop_def = f"{prop['name']}: {prop['type']}"
                if prop.get("unique"):
                    prop_def += " UNIQUE"
                if prop.get("required"):
                    prop_def = prop_def.replace(")", ")")
                props.append(prop_def)
            
            cypher += f"CREATE CONSTRAINT FOR (n:{node['label']}) REQUIRE n.{node['identifying_property']} IS UNIQUE;\n"
        
        return cypher

ecommerce_model = {
    "nodes": [
        {
            "label": "Customer",
            "properties": [
                {"name": "customer_id", "type": "STRING"},
                {"name": "name", "type": "STRING"},
                {"name": "email", "type": "STRING"},
                {"name": "signup_date", "type": "DATE"}
            ]
        },
        {
            "label": "Product",
            "properties": [
                {"name": "product_id", "type": "STRING"},
                {"name": "name", "type": "STRING"},
                {"name": "category", "type": "STRING"},
                {"name": "price", "type": "FLOAT"}
            ]
        },
        {
            "label": "Order",
            "properties": [
                {"name": "order_id", "type": "STRING"},
                {"name": "total", "type": "FLOAT"},
                {"name": "status", "type": "STRING"},
                {"name": "created_at", "type": "DATETIME"}
            ]
        }
    ],
    "relationships": [
        {
            "name": "PURCHASED",
            "from": "Customer",
            "to": "Product",
            "properties": [
                {"name": "purchase_date", "type": "DATE"},
                {"name": "quantity", "type": "INT"}
            ]
        },
        {
            "name": "PLACED",
            "from": "Customer",
            "to": "Order"
        },
        {
            "name": "CONTAINS",
            "from": "Order",
            "to": "Product",
            "properties": [
                {"name": "quantity", "type": "INT"},
                {"name": "price_at_purchase", "type": "FLOAT"}
            ]
        },
        {
            "name": "SIMILAR_TO",
            "from": "Product",
            "to": "Product",
            "properties": [
                {"name": "score", "type": "FLOAT"}
            ]
        }
    ]
}

class GraphAlgorithms:
    @staticmethod
    def setup_graph_projection(graph_name: str, neo4j: Neo4jConnection):
        queries = {
            "project_social": f"""
                CALL gds.graph.project(
                    '{graph_name}',
                    'User',
                    {{
                        KNOWS: {{
                            type: 'KNOWS',
                            properties: 'weight'
                        }}
                    }}
                )
            """,
            
            "project_e commerce": f"""
                CALL gds.graph.project(
                    '{graph_name}',
                    ['Customer', 'Product'],
                    {{
                        PURCHASED: {{
                            type: 'PURCHASED',
                            properties: 'weight'
                        }},
                        SIMILAR_TO: {{
                            type: 'SIMILAR_TO',
                            properties: 'score'
                        }}
                    }}
                )
            """,
            
            "project_knowledge": f"""
                CALL gds.graph.project(
                    '{graph_name}',
                    ['Entity', 'Concept'],
                    '*'
                )
            """
        }
        
        return queries
    
    @staticmethod
    def pagerank(neo4j: Neo4jConnection) -> List[Dict]:
        query = """
            CALL gds.pageRank.stream('myGraph')
            YIELD nodeId, score
            RETURN gds.util.asNode(nodeId).name AS name, score
            ORDER BY score DESC
            LIMIT 20
        """
        return neo4j.execute(query)
    
    @staticmethod
    def community_detection(neo4j: Neo4jConnection) -> List[Dict]:
        query = """
            CALL gds.labelPropagation.stream('myGraph')
            YIELD nodeId, communityId
            RETURN communityId, 
                   collect(gds.util.asNode(nodeId).name) AS members
            ORDER BY size(members) DESC
        """
        return neo4j.execute(query)
    
    @staticmethod
    def node_similarity(neo4j: Neo4jConnection, min_similarity: float = 0.5) -> List[Dict]:
        query = """
            CALL gds.nodeSimilarity.stream('myGraph')
            YIELD node1, node2, similarity
            WHERE similarity > $minSimilarity
            RETURN gds.util.asNode(node1).name AS item1,
                   gds.util.asNode(node2).name AS item2,
                   similarity
            ORDER BY similarity DESC
            LIMIT 50
        """
        return neo4j.execute(query, {"minSimilarity": min_similarity})
    
    @staticmethod
    def shortest_path(neo4j: Neo4jConnection, from_node: str, to_node: str) -> Dict:
        query = """
            MATCH (start {name: $from}), (end {name: $to})
            CALL gds.shortestPath.dijkstra.stream(
                'myGraph',
                {{ startNode: start, endNode: end, relationshipWeightProperty: 'weight' }}
            )
            YIELD path, pathWeight
            RETURN path, pathWeight
        """
        return neo4j.execute_single(query, {"from": from_node, "to": to_node})
    
    @staticmethod
    def betweenness_centrality(neo4j: Neo4jConnection) -> List[Dict]:
        query = """
            CALL gds.betweenness.stream('myGraph')
            YIELD nodeId, score
            RETURN gds.util.asNode(nodeId).name AS name, score
            ORDER BY score DESC
            LIMIT 20
        """
        return neo4j.execute(query)

class FraudDetectionGraph:
    def __init__(self, neo4j: Neo4jConnection):
        self.db = neo4j
    
    def detect_connected_fraud(self, entity_id: str) -> Dict:
        query = """
            MATCH (suspicious {entityId: $entityId})-[*1..3]-(fraud)
            WHERE fraud:Account AND fraud.isFraudulent = true
            RETURN collect(DISTINCT fraud.entityId) AS connectedFraud,
                   length(*) AS distance
            ORDER BY distance
        """
        return self.db.execute_single(query, {"entityId": entity_id})
    
    def find_ring_formation(self, min_members: int = 3) -> List[Dict]:
        query = """
            MATCH (a:Account)-[:TRANSFERRED]->(b:Account)-[:TRANSFERRED]->(c:Account)
            WHERE a <> c
            AND exists((c:Account)-[:TRANSFERRED]->(a:Account))
            WITH a, b, c, 
                 [a.accountId, b.accountId, c.accountId] AS ring
            RETURN ring, size(ring) AS ringSize
            LIMIT 20
        """
        return self.db.execute(query, {"minMembers": min_members})
    
    def calculate_entity_risk_score(self, entity_id: str) -> Dict:
        query = """
            MATCH (entity {entityId: $entityId})
            OPTIONAL MATCH (entity)-[:HAS_PHONE]->(phone)
            OPTIONAL MATCH (entity)-[:HAS_EMAIL]->(email)
            OPTIONAL MATCH (entity)-[:HAS_ADDRESS]->(address)
            OPTIONAL MATCH (entity)-[:SHARED_IP]->(ip)<-[:SHARED_IP]-(other:Account)
            WHERE other.isFraudulent = true
            
            WITH entity, 
                 count(DISTINCT phone) AS phoneCount,
                 count(DISTINCT email) AS emailCount,
                 count(DISTINCT address) AS addressCount,
                 count(DISTINCT other) AS fraudConnections
            
            RETURN entity.entityId AS entityId,
                   phoneCount + emailCount + addressCount AS uniqueIdentifiers,
                   fraudConnections AS fraudRiskSignals,
                   CASE 
                       WHEN fraudConnections > 2 THEN 'HIGH'
                       WHEN fraudConnections > 0 THEN 'MEDIUM'
                       ELSE 'LOW'
                   END AS riskLevel
        """
        return self.db.execute_single(query, {"entityId": entity_id})

class RecommendationEngine:
    def __init__(self, neo4j: Neo4jConnection):
        self.db = neo4j
    
    def recommend_products(self, customer_id: str, limit: int = 10) -> List[Dict]:
        query = """
            MATCH (customer:Customer {customerId: $customerId})
            
            // Products bought by similar customers
            MATCH (customer)-[:PURCHASED]->(purchased:Product)<-[:PURCHASED]-(similar:Customer)
            WHERE customer <> similar
            
            // Weight by similarity
            WITH customer, purchased, similar,
                 count(DISTINCT similar) AS similarityScore
            
            // Exclude already purchased
            WHERE NOT (customer)-[:PURCHASED]->(purchased)
            
            // Aggregate and rank
            WITH purchased, sum(similarityScore) AS score
            ORDER BY score DESC
            LIMIT $limit
            
            RETURN purchased.productId AS productId,
                   purchased.name AS name,
                   purchased.category AS category,
                   score AS recommendationScore
        """
        return self.db.execute(query, {"customerId": customer_id, "limit": limit})
    
    def recommend_complementary_products(self, product_id: str, limit: int = 10) -> List[Dict]:
        query = """
            MATCH (product:Product {productId: $productId})
            
            // Products frequently bought together
            MATCH (product)<-[:CONTAINS]-(:Order)-[:CONTAINS]->(together:Product)
            WHERE product <> together
            
            WITH together,
                 count(*) AS purchaseFrequency
            ORDER BY purchaseFrequency DESC
            LIMIT $limit
            
            RETURN together.productId AS productId,
                   together.name AS name,
                   purchaseFrequency
        """
        return self.db.execute(query, {"productId": product_id, "limit": limit})

### Category-Based Recommendations

def get_product_recommendations(user_id, limit=5):
    """Get product recommendations based on user behavior"""

    session = get_session()

    result = session.run("""
        MATCH (user:User {id: $user_id})-[:PURCHASED]->(product:Product)
        MATCH (product)-[:IN_CATEGORY]->(category:Category)
        MATCH (category)<-[:IN_CATEGORY]-(recommended:Product)
        WHERE NOT (user)-[:PURCHASED]->(recommended)
        RETURN recommended.name as product,
               COUNT(*) as score
        ORDER BY score DESC
        LIMIT $limit
    """, user_id=user_id, limit=limit)

    recommendations = [record for record in result]
    session.close()

    return recommendations

def get_collaborative_recommendations(user_id, limit=5):
    """Get recommendations from similar users"""

    session = get_session()

    result = session.run("""
        MATCH (user:User {id: $user_id})-[:PURCHASED]->(product:Product)
        MATCH (similar_user:User)-[:PURCHASED]->(product)
        WHERE similar_user.id <> user.id
        MATCH (similar_user)-[:PURCHASED]->(recommended:Product)
        WHERE NOT (user)-[:PURCHASED]->(recommended)
        RETURN recommended.name as product,
               COUNT(*) as score
        ORDER BY score DESC
        LIMIT $limit
    """, user_id=user_id, limit=limit)

    recommendations = [record for record in result]
    session.close()

    return recommendations

class NetworkAnalysis:
    def __init__(self, neo4j: Neo4jConnection):
        self.db = neo4j
    
    def identify_influencers(self, min_connections: int = 100) -> List[Dict]:
        query = """
            CALL gds.degree.stream('socialGraph')
            YIELD nodeId, score
            WHERE score > $minConnections
            RETURN gds.util.asNode(nodeId).name AS user,
                   score AS connections,
                   'INFLUENCER' AS classification
            ORDER BY score DESC
            LIMIT 50
        """
        return self.db.execute(query, {"minConnections": min_connections})
    
    def find_communities(self) -> List[Dict]:
        query = """
            CALL gds.louvain.stream('socialGraph')
            YIELD nodeId, communityId
            WITH communityId, collect(gds.util.asNode(nodeId).name) AS members
            WHERE size(members) > 10
            RETURN communityId, members
            ORDER BY size(members) DESC
        """
        return self.db.execute(query)
    
    def analyze_information_flow(self, start_node: str, max_depth: int = 3) -> Dict:
        query = """
            MATCH (start {name: $startNode})
            CALL gds.bfs.stream('socialGraph', {startNode: start})
            YIELD path
            WITH path, length(path) AS depth
            WHERE depth <= $maxDepth
            RETURN nodes(path) AS chain, depth
            ORDER BY depth
        """
        return self.db.execute(query, {"startNode": start_node, "maxDepth": max_depth})

class GraphRelationalIntegration:
    @staticmethod
    def export_to_graph(neo4j: Neo4jConnection, connection, table_mapping: Dict):
        for table, mapping in table_mapping.items():
            nodes_query = f"""
                MATCH (n:{mapping['label']})
                RETURN n.{mapping['id_property']} AS id
            """
            existing = set(r["id"] for r in neo4j.execute(nodes_query))
            
            cursor = connection.cursor()
            cursor.execute(f"SELECT * FROM {table}")
            
            for row in cursor.fetchall():
                record_id = row[mapping['id_column']]
                if record_id not in existing:
                    properties = {
                        col: value 
                        for col, value in zip(cursor.description, row)
                        if col != mapping['id_column']
                    }
                    neo4j.execute(
                        f"CREATE (n:{mapping['label']} $props)",
                        {"props": properties}
                    )
    
    @staticmethod
    def import_from_graph(neo4j: Neo4jConnection, connection, query: str, table: str):
        records = neo4j.execute(query)
        
        if records:
            columns = list(records[0].keys())
            placeholders = ", ".join(["%s"] * len(columns))
            
            insert_query = f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({placeholders})"
            
            cursor = connection.cursor()
            cursor.executemany(insert_query, [tuple(r.values()) for r in records])
            connection.commit()

class GraphKafkaIntegration:
    def __init__(self, neo4j: Neo4jConnection, kafka_bootstrap_servers: str):
        self.neo4j = neo4j
        self.kafka_config = {
            "bootstrap.servers": kafka_bootstrap_servers,
            "group.id": "graph-connector"
        }
    
    def stream_node_events(self, topic: str, label: str):
        from kafka import KafkaConsumer
        
        consumer = KafkaConsumer(topic, **self.kafka_config)
        
        for message in consumer:
            event = json.loads(message.value)
            
            if event["operation"] == "CREATE":
                self.neo4j.execute(
                    f"CREATE (n:{label} $props)",
                    {"props": event["properties"]}
                )
            elif event["operation"] == "UPDATE":
                self.neo4j.execute(
                    f"MATCH (n:{label} {{id: $id}}) SET n += $props",
                    {"id": event["id"], "props": event["properties"]}
                )
            elif event["operation"] == "DELETE":
                self.neo4j.execute(
                    f"MATCH (n:{label} {{id: $id}}) DETACH DELETE n",
                    {"id": event["id"]}
                )
    
    def stream_relationship_events(self, topic: str):
        from kafka import KafkaConsumer
        
        consumer = KafkaConsumer(topic, **self.kafka_config)
        
        for message in consumer:
            event = json.loads(message.value)
            
            if event["operation"] == "CREATE":
                self.neo4j.execute(
                    f"""
                    MATCH (a {{id: $fromId}}), (b {{id: $toId}})
                    CREATE (a)-[r:{event['type']} $props]->(b)
                    """,
                    {"fromId": event["from"], "toId": event["to"], "props": event.get("properties", {})}
                )

class GraphIndexing:
    @staticmethod
    def create_indexes(neo4j: Neo4jConnection):
        indexes = [
            "CREATE INDEX person_name IF NOT EXISTS FOR (p:Person) ON (p.name)",
            "CREATE INDEX person_email IF NOT EXISTS FOR (p:Person) ON (p.email)",
            "CREATE INDEX product_category IF NOT EXISTS FOR (p:Product) ON (p.category)",
            "CREATE INDEX product_price IF NOT EXISTS FOR (p:Product) ON (p.price)",
            "CREATE INDEX order_date IF NOT EXISTS FOR (o:Order) ON (o.createdAt)",
            
            // Composite indexes
            "CREATE INDEX person_name_email IF NOT EXISTS FOR (p:Person) ON (p.name, p.email)",
            
            // Full-text indexes
            """CREATE FULLTEXT INDEX personSearch IF NOT EXISTS 
               FOR (p:Person) ON [p.name, p.bio]"""
        ]
        
        for idx in indexes:
            neo4j.execute(idx)
    
    @staticmethod
    def create_constraints(neo4j: Neo4jConnection):
        constraints = [
            "CREATE CONSTRAINT personId IF NOT EXISTS FOR (p:Person) REQUIRE p.id IS UNIQUE",
            "CREATE CONSTRAINT productId IF NOT EXISTS FOR (p:Product) REQUIRE p.id IS UNIQUE",
            "CREATE CONSTRAINT orderId IF NOT EXISTS FOR (o:Order) REQUIRE o.id IS UNIQUE"
        ]
        
        for constraint in constraints:
            neo4j.execute(constraint)

class GraphQueryOptimization:
    @staticmethod
    def explain_query(neo4j: Neo4jConnection, query: str):
        return neo4j.execute(f"EXPLAIN {query}")
    
    @staticmethod
    def profile_query(neo4j: Neo4jConnection, query: str):
        return neo4j.execute(f"PROFILE {query}")
    
    @staticmethod
    def optimize_pattern_matching():
        return {
            "use_relationship_direction": "Always specify direction when possible",
            "avoid_star_patterns": "Use specific relationship types instead of *",
            "limit_results_early": "Use LIMIT before expensive operations",
            "use_parameters": "Pass values as parameters, not literals",
            "create_indexes": "Index frequently queried properties",
            "use_projection": "Return only needed properties"
        }

def create_arangodb_indexes():
    """Create indexes in ArangoDB"""

    people_collection = db.collection('people')

    # Create hash index
    people_collection.add_hash_index(fields=['name'], unique=False)

    # Create skiplist index
    people_collection.add_skiplist_index(fields=['age'], unique=False)

    # Create fulltext index
    people_collection.add_fulltext_index(fields=['name'], min_length=3)

from arango import ArangoClient

class ArangoDBGraph:
    def __init__(self, host: str, username: str, password: str):
        self.client = ArangoClient(host=host)
        self.db = self.client.db(username, password, verify=True)
    
    def create_vertex_collection(self, name: str):
        if not self.db.has_collection(name):
            self.db.create_collection(name)
    
    def create_edge_collection(self, name: str, from_col: str, to_col: str):
        edge_def = {
            "name": name,
            "from": [from_col],
            "to": [to_col]
        }
        if not self.db.has_collection(name):
            self.db.create_collection(name, edge=True, **edge_def)
    
    def insert_vertex(self, collection: str, data: Dict):
        return self.db.collection(collection).insert(data)
    
    def insert_edge(self, collection: str, from_key: str, to_key: str, data: Dict):
        edge = {
            "_from": f"{collection}/{from_key}",
            "_to": f"{collection}/{to_key}",
            **data
        }
        return self.db.collection(collection).insert(edge)
    
    def aql_query(self, query: str, bind_vars: Dict = None):
        return self.db.aql.execute(query, bind_vars=bind_vars)

def create_arangodb_graph():
    """Create graph in ArangoDB"""

    # Create collections
    if not db.has_collection('people'):
        db.create_collection('people')

    if not db.has_collection('relationships'):
        db.create_collection('relationships', edge=True)

    people_collection = db.collection('people')
    relationships_collection = db.collection('relationships')

    # Insert people
    people_collection.insert_many([
        {'_key': 'alice', 'name': 'Alice', 'age': 30},
        {'_key': 'bob', 'name': 'Bob', 'age': 28},
        {'_key': 'charlie', 'name': 'Charlie', 'age': 32},
        {'_key': 'diana', 'name': 'Diana', 'age': 29}
    ])

    # Insert relationships
    relationships_collection.insert_many([
        {'_from': 'people/alice', '_to': 'people/bob', 'type': 'knows', 'since': 2020},
        {'_from': 'people/bob', '_to': 'people/charlie', 'type': 'knows', 'since': 2019},
        {'_from': 'people/alice', '_to': 'people/diana', 'type': 'knows', 'since': 2021}
    ])

    print("ArangoDB graph created")

def aql_find_friends_of_friends(person_name):
    """Find friends of friends using AQL"""

    aql = """
        FOR person IN people
            FILTER person.name == @name
            FOR friend IN 1..1 OUTBOUND person relationships
                FOR fof IN 1..1 OUTBOUND friend relationships
                    FILTER fof._key != person._key
                    RETURN DISTINCT fof.name
    """

    cursor = db.aql.execute(aql, bind_vars={'name': person_name})
    return [doc for doc in cursor]

def aql_shortest_path(start_name, end_name):
    """Find shortest path using AQL"""

    aql = """
        FOR v, e, p IN 1..10 OUTBOUND
            CONCAT('people/', @start) relationships
            FILTER v.name == @end
            RETURN {
                path: [node IN p.vertices[*] RETURN node.name],
                distance: LENGTH(p.edges)
            }
            LIMIT 1
    """

    cursor = db.aql.execute(aql, bind_vars={
        'start': start_name,
        'end': end_name
    })

    return [doc for doc in cursor]

def aql_graph_analytics():
    """Perform graph analytics"""

    aql = """
        FOR person IN people
            LET connections = LENGTH(
                FOR rel IN relationships
                    FILTER rel._from == person._id
                    RETURN rel
            )
            RETURN {
                name: person.name,
                connections: connections
            }
            ORDER BY connections DESC
    """

    cursor = db.aql.execute(aql)
    return [doc for doc in cursor]

best_practices = {
    "naming_conventions": [
        "Use PascalCase for labels: Person, OrderItem",
        "Use UPPER_SNAKE for relationship types: KNOWS, PURCHASED",
        "Use camelCase for properties: firstName, createdAt"
    ],
    
    "relationship_design": [
        "Use directed relationships by default",
        "Create bidirectional relationships explicitly when needed",
        "Use relationship properties instead of node properties for attributes that vary per relationship",
        "Prefer few relationship types with properties over many types"
    ],
    
    "performance": [
        "Create indexes on frequently queried properties",
        "Use constraints to enforce uniqueness",
        "Avoid patterns that scan entire graphs",
        "Use projection to limit returned data"
    ],
    
    "modeling_patterns": [
        "Use intermediate nodes for many-to-many relationships with attributes",
        "Use relationship types to categorize similar connections",
        "Consider time-based relationships for historical data",
        "Denormalize for read performance when appropriate"
    ]
}

operational_best_practices = {
    "backup": [
        "Schedule regular backups using neo4j-admin backup",
        "Test restore procedures regularly",
        "Consider point-in-time recovery for critical systems"
    ],
    
    "monitoring": [
        "Monitor query latency and throughput",
        "Track slow queries with profiling",
        "Set up alerts for unusual patterns"
    ],
    
    "security": [
        "Enable SSL for Bolt and HTTP protocols",
        "Use role-based access control",
        "Implement query result size limits",
        "Audit sensitive operations"
    ],
    
    "scaling": [
        "Understand read vs write patterns",
        "Use causal clustering for high availability",
        "Consider horizontal scaling with Causal Cluster",
        "Use appropriate instance sizes for workload"
    ]
}