groc-prog / pyneo4j-ogm Goto Github PK

import uuid

from pyneo4j_ogm import (
    Pyneo4jClient,
    NodeModel,
    RelationshipProperty,
    RelationshipPropertyCardinality,
    RelationshipPropertyDirection,
    RelationshipModel,
)


class Default(RelationshipModel):
    class Settings:
        type = "DEFAULT"


class Resource(NodeModel):
    ...


class Company(NodeModel):
    resources: RelationshipProperty["Resource", Default] = RelationshipProperty(
        target_model=Resource,
        relationship_model=Default,
        direction=RelationshipPropertyDirection.OUTGOING,
        cardinality=RelationshipPropertyCardinality.ZERO_OR_MORE,
        allow_multiple=False,
    )


class User(NodeModel):
    companies: RelationshipProperty["Company", Default] = RelationshipProperty(
        target_model="Company",
        relationship_model=Default,
        direction=RelationshipPropertyDirection.OUTGOING,
        cardinality=RelationshipPropertyCardinality.ZERO_OR_MORE,
        allow_multiple=False,
    )


async def main():
    client = Pyneo4jClient()
    await client.connect(
        uri="neo4j://localhost:7687", auth=("neo4j", "password"), database="neo4j"
    )
    await client.register_models([Default, Resource, Company, User])
    user = User()
    await user.create()
    resource = Resource()
    await resource.create()
    company = Company()
    await company.create()
    await user.companies.connect(company)
    await company.resources.connect(resource)
    reachable_resources = await user.find_connected_nodes(  # here TypeError is raised
        {
            "$maxHops": "*",
            "$node": {
                "$labels": resource._settings.labels,
            },
        }
    )
    client.close()
    print(reachable_resources)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

class VehicleNode(NodeModel):
    engine: str


class CarNode(VehicleNode):
    has_windows: bool


class Consumed(RelationshipModel):
    liters_consumed: int


class GasStationNode(NodeModel):
    volume_in_litres: int
    clients: RelationshipProperty[
        VehicleNode, Consumed
    ] = RelationshipProperty(
        target_model=VehicleNode,
        relationship_model=Consumed,
        direction=RelationshipPropertyDirection.OUTGOING,
        cardinality=RelationshipPropertyCardinality.ZERO_OR_MORE,
        allow_multiple=True
    )

client: Pyneo4jClient # Already initialized
await client.register_models([VehicleNode, CarNode, GasStationNode, Consumed])
    
car = await CarNode(engine='v8', has_windows=True).create()
gas_station = await GasStationNode(volume_in_litres=6000).create()

# Trying to connect them by basic relationship
await gas_station.clients.connect(car, { 'liters_consumed': 50 })

self = RelationshipProperty(target_model_name=VehicleNode, relationship_model=Consumed, direction=OUTGOING, cardinality=ZERO_OR_MORE, allow_multiple=True)
nodes = CarNode(element_id=4:218fdb95-3060-4d65-ab59-de3bc9a5722f:1, destroyed=False)

    def _ensure_alive(self, nodes: Union[T, List[T]]) -> None:
        """
        Ensures that the provided nodes are alive.
    
        Args:
            nodes (T | List[T]): Nodes to check for hydration and alive.
    
        Raises:
            InstanceNotHydrated: Raised if a node is not hydrated yet.
            InstanceDestroyed: Raised if a node has been marked as destroyed.
        """
        nodes_to_check = nodes if isinstance(nodes, list) else [nodes]
    
        for node in nodes_to_check:
            logger.debug(
                "Checking if node %s is alive and of correct type",
                getattr(node, "_element_id", None),
            )
            if getattr(node, "_element_id", None) is None or getattr(node, "_id", None) is None:
                raise InstanceNotHydrated()
    
            if getattr(node, "_destroyed", True):
                raise InstanceDestroyed()
    
            if cast(Type[T], self._target_model).__name__ != node.__class__.__name__:
>               raise InvalidTargetNode(
                    expected_type=cast(Type[T], self._target_model).__name__,
                    actual_type=node.__class__.__name__,
                )
E               pyneo4j_ogm.exceptions.InvalidTargetNode: Expected target node to be of type VehicleNode, but got CarNode