Advanced¶

# /// script
# dependencies = [
#   "graphinate",
# ]
# ///
import operator
from typing import Annotated

import graphinate
from graphinate import ParentId
from graphinate.builders import NetworkxBuilder
from graphinate.engine import EdgeEvent, GraphEngine, GraphObserver, NodeEvent


class AuditObserver(GraphObserver):
    """Custom observer that collects streaming statistics without building a graph."""

    def __init__(self) -> None:
        self.model_name: str = ''
        self.nodes_by_type: dict[str, int] = {}
        self.edge_count: int = 0

    def on_start(self, model_name: str) -> None:
        self.model_name = model_name
        self.nodes_by_type.clear()
        self.edge_count = 0

    def on_node(self, event: NodeEvent) -> None:
        self.nodes_by_type[event.type] = self.nodes_by_type.get(event.type, 0) + 1

    def on_edge(self, event: EdgeEvent) -> None:
        self.edge_count += 1

    def on_finish(self) -> None:
        print(f'Audit complete ({self.model_name}): {self.nodes_by_type}, total edges={self.edge_count}')


def create_cloud_topology_model() -> graphinate.GraphModel:
    cloud_model = graphinate.model(name='Cloud Topology')

    @cloud_model.node(type_='region', key=operator.itemgetter('id'), label=operator.itemgetter('name'))
    def regions():
        yield {'id': 'us-east', 'name': 'US East'}
        yield {'id': 'eu-west', 'name': 'EU West'}

    @cloud_model.node(type_='vpc', parent_type='region', key=operator.itemgetter('id'))
    def vpcs(region_id: Annotated[str, ParentId('region')]):
        yield {'id': f'{region_id}-vpc-1'}

    @cloud_model.node(
        type_='service',
        parent_type='vpc',
        key=operator.itemgetter('id'),
        unique=True,
    )
    def services(
        region_id: Annotated[str, ParentId('region')],
        vpc_id: Annotated[str, ParentId('vpc')],
    ):
        yield {'id': f'{vpc_id}-api', 'region': region_id}
        yield {'id': f'{vpc_id}-db', 'region': region_id}

    @cloud_model.edge(type_='peering')
    def peering_links():
        yield {'source': 'us-east-vpc-1-api', 'target': 'eu-west-vpc-1-api'}

    return cloud_model


if __name__ == '__main__':
    topology_model = create_cloud_topology_model()
    audit_observer = AuditObserver()
    nx_builder = NetworkxBuilder(topology_model)

    # Execute a single traversal pass that notifies both observers simultaneously
    engine = GraphEngine(topology_model)
    engine.subscribe(audit_observer)
    engine.subscribe(nx_builder)
    engine.run()
    print('NetworkX Graph summary:', nx_builder._graph)

# /// script
# dependencies = [
#   "graphinate",
# ]
# ///
import asyncio
import operator
from typing import Annotated

import graphinate
from graphinate import ParentId, abuild
from graphinate.builders import D3Builder, MermaidBuilder, NetworkxBuilder


def create_async_service_model() -> graphinate.GraphModel:
    """Create a model mixing async and synchronous node and edge suppliers."""
    service_model = graphinate.model(name='Hybrid Async/Sync Service Mesh')

    # 1. Async generator (async def + yield)
    @service_model.node(type_='cluster', key=operator.itemgetter('id'), label=operator.itemgetter('name'))
    async def clusters():
        await asyncio.sleep(0.01)
        yield {'id': 'prod', 'name': 'Production Cluster'}

    # 2. Async coroutine returning a collection (async def + return)
    @service_model.node(type_='service', parent_type='cluster', key=operator.itemgetter('id'))
    async def services(cluster_id: Annotated[str, ParentId('cluster')]):
        await asyncio.sleep(0.01)
        return [
            {'id': f'{cluster_id}-gateway'},
            {'id': f'{cluster_id}-auth'},
            {'id': f'{cluster_id}-billing'},
        ]

    # 3. Synchronous generator (def + yield) nested under an async parent
    @service_model.node(
        type_='endpoint',
        parent_type='service',
        key=operator.itemgetter('id'),
        unique=True,
    )
    def endpoints(service_id: Annotated[str, ParentId('service')]):
        if service_id.endswith('gateway'):
            yield {'id': f'{service_id}:/api/v1'}
        elif service_id.endswith('auth'):
            yield {'id': f'{service_id}:/oauth/token'}

    # 4. Async edge generator
    @service_model.edge(type_='calls', label=operator.itemgetter('protocol'))
    async def service_calls():
        await asyncio.sleep(0.01)
        yield {'source': 'prod-gateway', 'target': 'prod-auth', 'protocol': 'gRPC'}
        yield {'source': 'prod-gateway', 'target': 'prod-billing', 'protocol': 'HTTP/2'}

    # 5. Synchronous edge generator in the same model
    @service_model.edge(type_='routes_to')
    def endpoint_routes():
        yield {'source': 'prod-gateway:/api/v1', 'target': 'prod-auth:/oauth/token'}

    return service_model


async def main() -> None:
    model = create_async_service_model()

    nx_graph = await abuild(NetworkxBuilder, model)
    print('Hybrid Async/Sync NetworkX Graph:', nx_graph)

    d3_graph = await D3Builder(model).abuild()
    print(f"D3 Nodes: {len(d3_graph['nodes'])}, Links: {len(d3_graph['links'])}")

    mermaid_diagram = await MermaidBuilder(model).abuild()
    print('\nMermaid Diagram:\n' + mermaid_diagram)


if __name__ == '__main__':
    asyncio.run(main())

# /// script
# dependencies = [
#   "graphinate",
# ]
# ///
import graphinate
from graphinate import CompositionStrategy, SchemaMode
from graphinate.builders import GraphQLBuilder, NetworkxBuilder


def create_composed_model() -> graphinate.GraphModel:
    compute_model = graphinate.model(name='Compute')

    @compute_model.node(type_='service')
    def compute_services():
        yield 'api-gateway'
        yield 'worker'

    storage_model = graphinate.model(name='Storage')

    @storage_model.node(type_='database')
    def databases():
        yield 'postgres-primary'

    @storage_model.edge(type_='reads_from')
    def connections():
        yield {'source': 'api-gateway', 'target': 'postgres-primary'}
        yield {'source': 'worker', 'target': 'postgres-primary'}

    return compute_model.merge(
        storage_model,
        strategy=CompositionStrategy.MERGE,
        name='Platform Graph',
    )


if __name__ == '__main__':
    platform_model = create_composed_model()

    # 1. Dynamic Schema Mode (generates ServiceNode, DatabaseNode, services, databases)
    dynamic_schema = GraphQLBuilder(platform_model, mode=SchemaMode.DYNAMIC).build()
    dynamic_result = dynamic_schema.execute_sync(
        '{ services { label neighbors { type label } } databases { label } }'
    )
    print('Dynamic Schema Result:', dynamic_result.data)

    # 2. Static Schema Mode (uses a single GenericGraphNode and fixed root fields)
    static_schema = GraphQLBuilder(platform_model, mode=SchemaMode.STATIC).build()
    static_result = static_schema.execute_sync(
        '{ nodes { type label } edges { type label } }'
    )
    print('Static Schema Result:', static_result.data)

    # 3. Context-Injected Graph Execution (override the backing NetworkX graph per request)
    custom_graph = NetworkxBuilder(platform_model).build()
    custom_graph.graph['name'] = 'Context-Injected Snapshot'
    injected_result = static_schema.execute_sync(
        '{ graph { name nodeCount edgeCount } }',
        context_value={'graph': custom_graph},
    )
    print('Context-Injected Result:', injected_result.data)