Skip to content

gremlin-python-tutorial API Documentation

draw

Created on 2023-05-15

@author: jv

GremlinDraw

helper class to draw Gremlin Graphs via Graphviz

Source code in gremlin/draw.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class GremlinDraw:
    """
    helper class to draw Gremlin Graphs via Graphviz
    """

    def __init__(
        self, g: GraphTraversalSource, title: str, config: GremlinDrawConfig = None
    ):
        """
        constructor
        """
        self.g = g
        self.title = title
        if config is None:
            config = GremlinDrawConfig()
        self.config = config
        self.gviz: graphviz.Digraph = graphviz.Digraph(
            title, format=config.output_format
        )
        # keep track of the vertices and edges drawn
        self.v_drawn = {}
        self.e_drawn = {}

    def __as_label(self, head, body: str) -> str:
        """
        create a label from head and body separated by a dash
        with the configured width
        """
        # note the UTF-8 dash ...
        dash = "─" * self.config.dash_width
        label = f"{head}\n{dash}\n{body}"
        return label

    def get_vertex_properties(self, vertex: Vertex) -> list:
        """
        get the properties for a given vertex
        """
        # developer note: see https://github.com/apache/tinkerpop/blob/master/gremlin-python/src/main/python/gremlin_python/structure/graph.py#LL58C23-L58C23
        # has properties but these are not set as for gremlin-python 3.7.0

        # get the properties of the vertex (work around)
        kvp_list = list(next(self.g.V(vertex).element_map()).items())
        # non-property items are of type aenum
        properties = [item for item in kvp_list if not isinstance(item[0], Enum)]
        assert len(properties) == len(kvp_list) - 2  # ID and label are not properties
        if self.config.vertex_properties is not None:
            properties = [
                item for item in properties if item[0] in self.config.vertex_properties
            ]
        return properties

    def get_edge_properties(self, edge: Edge) -> list:
        # developer note: see https://github.com/apache/tinkerpop/blob/master/gremlin-python/src/main/python/gremlin_python/structure/graph.py#L66
        # when gremlin-python 3.7.0 is released, the following code might be improved (get the properties using edge.properties)
        # e_props=edge.properties
        # 2023-08-21: WF tested - but properties are not set ...
        # then, g can also be removed as a parameter
        # get the properties of the edge
        edge_t = self.g.E(edge)
        try:
            edge_map = edge_t.element_map().next()
            kvp_list = list(edge_map.items())
        except StopIteration:
            pass
            return []

        # Workaround, because the above line does not work due to inconsistencies / bugs in the gremlin-python library
        # kvp_list = [edge_element_map for edge_element_map in self.g.E().element_map().to_list() if edge_element_map[T.id] == edge.id][0].items()
        # non-property items are of type aenum
        properties = [item for item in kvp_list if not isinstance(item[0], Enum)]
        assert (
            len(properties) == len(kvp_list) - 4
        )  # ID, label, in, and out are not properties
        if self.config.edge_properties is not None:
            properties = [
                item for item in properties if item[0] in self.config.edge_properties
            ]
        return properties

    def draw_vertex(self, vertex: Vertex):
        """
        draw a single given vertex
        """
        # avoid drawing to many vertices
        if len(self.v_drawn) >= self.config.v_limit:
            return
        if vertex.id in self.v_drawn:
            return
        properties = self.get_vertex_properties(vertex)
        properties_label = "\n".join(f"{key}: {value}" for key, value in properties)
        head = f"{str(vertex.id)}\n{vertex.label}"
        body = f"{properties_label}"
        label = self.__as_label(head, body)
        # draw the vertex
        self.gviz.node(
            name=str(vertex.id),
            label=f"{label}",
            fillcolor=f"{self.config.fillcolor}",
            style="filled",
            fontname=f"{self.config.fontname}",
        )
        self.v_drawn[vertex.id] = vertex

    def draw_edge(self, edge: Edge, with_vertices: bool = True):
        """
        draw a single given edge
        """
        # avoid drawing to many vertices
        if len(self.e_drawn) >= self.config.e_limit:
            return
        if edge.id in self.e_drawn:
            return
        if with_vertices:
            self.draw_vertex(edge.inV)
            self.draw_vertex(edge.outV)
            pass
        properties = self.get_edge_properties(edge)
        properties_label = "\n".join(f"{key}: {value}" for key, value in properties)
        head = f"{str(edge.id)}\n{edge.label}"
        body = properties_label
        label = self.__as_label(head, body)
        # get the image of the edge by id
        in_vertex_id = edge.inV.id
        out_vertex_id = edge.outV.id

        # draw the edge
        self.gviz.edge(
            tail_name=str(out_vertex_id),
            head_name=str(in_vertex_id),
            label=f"{label}",
            style=f"setlinewidth({self.config.edge_line_width})",
            fontname=f"{self.config.fontname}",
        )
        self.e_drawn[edge.id] = edge

    def draw_g(self):
        # draw vertices
        vlist = self.g.V().to_list()
        vlist = vlist[: self.config.v_limit]

        for v in vlist:
            self.draw_vertex(v)

        # draw edges
        elist = self.g.E().to_list()
        elist = elist[: self.config.e_limit]

        for e in elist:
            self.draw_edge(e)

    def draw(self, gt: Union[GraphTraversal, Any]):
        # developer note: when moving the minimum supported version up to 3.10, the following code can be greatly improved by using match statements
        worklist: List[Any] = (
            gt.to_list()
            if isinstance(gt, GraphTraversal)
            else list(gt) if isinstance(gt, Iterable) else [gt]
        )

        while len(worklist) > 0:
            # move any vertices to the front of the worklist (draw them first)
            worklist = [item for item in worklist if not isinstance(item, Vertex)] + [
                item for item in worklist if isinstance(item, Vertex)
            ]

            result = worklist.pop(0)

            if isinstance(result, Vertex):
                self.draw_vertex(result)
            elif isinstance(result, Edge):
                self.draw_edge(result)
            elif isinstance(result, Path):
                for item in result.objects:
                    worklist.append(item)
            elif isinstance(result, dict):
                if T.id in result:
                    # check if the id is a vertex or an edge
                    if self.g.V(result[T.id]).hasNext():
                        self.draw_vertex(next(self.g.V(result[T.id])))
                    elif self.g.E(result[T.id]).hasNext():
                        self.draw_edge(self.g.E(result[T.id]).next())
                    else:
                        # raise Exception("id not found")
                        pass  # silent skip
                else:
                    # raise Exception("id not found")
                    pass  # silent skip
            else:
                # raise Exception(f"unknown type: {type(result)}")
                pass  # silent skip

    @staticmethod
    def show(
        g: GraphTraversalSource,
        title: str = "Gremlin",
        v_limit: int = 10,
        e_limit: int = 10,
    ) -> graphviz.Digraph:
        """
        draw the given graph
        """
        gd = GremlinDraw(g=g, title=title)
        gd.config.v_limit = v_limit
        gd.config.e_limit = e_limit
        gd.draw_g()
        return gd.gviz

    @staticmethod
    def show_graph_traversal(
        g: GraphTraversalSource, gt: Union[GraphTraversal, Any], title: str = "Gremlin"
    ) -> graphviz.Digraph:
        """
        draw the given graph traversal
        """
        gd = GremlinDraw(g=g, title=title)
        gd.draw(gt)
        return gd.gviz

__as_label(head, body)

create a label from head and body separated by a dash with the configured width

Source code in gremlin/draw.py 
60
61
62
63
64
65
66
67
68
def __as_label(self, head, body: str) -> str:
    """
    create a label from head and body separated by a dash
    with the configured width
    """
    # note the UTF-8 dash ...
    dash = "─" * self.config.dash_width
    label = f"{head}\n{dash}\n{body}"
    return label

__init__(g, title, config=None)

constructor

Source code in gremlin/draw.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self, g: GraphTraversalSource, title: str, config: GremlinDrawConfig = None
):
    """
    constructor
    """
    self.g = g
    self.title = title
    if config is None:
        config = GremlinDrawConfig()
    self.config = config
    self.gviz: graphviz.Digraph = graphviz.Digraph(
        title, format=config.output_format
    )
    # keep track of the vertices and edges drawn
    self.v_drawn = {}
    self.e_drawn = {}

draw_edge(edge, with_vertices=True)

draw a single given edge

Source code in gremlin/draw.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def draw_edge(self, edge: Edge, with_vertices: bool = True):
    """
    draw a single given edge
    """
    # avoid drawing to many vertices
    if len(self.e_drawn) >= self.config.e_limit:
        return
    if edge.id in self.e_drawn:
        return
    if with_vertices:
        self.draw_vertex(edge.inV)
        self.draw_vertex(edge.outV)
        pass
    properties = self.get_edge_properties(edge)
    properties_label = "\n".join(f"{key}: {value}" for key, value in properties)
    head = f"{str(edge.id)}\n{edge.label}"
    body = properties_label
    label = self.__as_label(head, body)
    # get the image of the edge by id
    in_vertex_id = edge.inV.id
    out_vertex_id = edge.outV.id

    # draw the edge
    self.gviz.edge(
        tail_name=str(out_vertex_id),
        head_name=str(in_vertex_id),
        label=f"{label}",
        style=f"setlinewidth({self.config.edge_line_width})",
        fontname=f"{self.config.fontname}",
    )
    self.e_drawn[edge.id] = edge

draw_vertex(vertex)

draw a single given vertex

Source code in gremlin/draw.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def draw_vertex(self, vertex: Vertex):
    """
    draw a single given vertex
    """
    # avoid drawing to many vertices
    if len(self.v_drawn) >= self.config.v_limit:
        return
    if vertex.id in self.v_drawn:
        return
    properties = self.get_vertex_properties(vertex)
    properties_label = "\n".join(f"{key}: {value}" for key, value in properties)
    head = f"{str(vertex.id)}\n{vertex.label}"
    body = f"{properties_label}"
    label = self.__as_label(head, body)
    # draw the vertex
    self.gviz.node(
        name=str(vertex.id),
        label=f"{label}",
        fillcolor=f"{self.config.fillcolor}",
        style="filled",
        fontname=f"{self.config.fontname}",
    )
    self.v_drawn[vertex.id] = vertex

get_vertex_properties(vertex)

get the properties for a given vertex

Source code in gremlin/draw.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def get_vertex_properties(self, vertex: Vertex) -> list:
    """
    get the properties for a given vertex
    """
    # developer note: see https://github.com/apache/tinkerpop/blob/master/gremlin-python/src/main/python/gremlin_python/structure/graph.py#LL58C23-L58C23
    # has properties but these are not set as for gremlin-python 3.7.0

    # get the properties of the vertex (work around)
    kvp_list = list(next(self.g.V(vertex).element_map()).items())
    # non-property items are of type aenum
    properties = [item for item in kvp_list if not isinstance(item[0], Enum)]
    assert len(properties) == len(kvp_list) - 2  # ID and label are not properties
    if self.config.vertex_properties is not None:
        properties = [
            item for item in properties if item[0] in self.config.vertex_properties
        ]
    return properties

show(g, title='Gremlin', v_limit=10, e_limit=10) staticmethod

draw the given graph

Source code in gremlin/draw.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@staticmethod
def show(
    g: GraphTraversalSource,
    title: str = "Gremlin",
    v_limit: int = 10,
    e_limit: int = 10,
) -> graphviz.Digraph:
    """
    draw the given graph
    """
    gd = GremlinDraw(g=g, title=title)
    gd.config.v_limit = v_limit
    gd.config.e_limit = e_limit
    gd.draw_g()
    return gd.gviz

show_graph_traversal(g, gt, title='Gremlin') staticmethod

draw the given graph traversal

Source code in gremlin/draw.py 
243
244
245
246
247
248
249
250
251
252
@staticmethod
def show_graph_traversal(
    g: GraphTraversalSource, gt: Union[GraphTraversal, Any], title: str = "Gremlin"
) -> graphviz.Digraph:
    """
    draw the given graph traversal
    """
    gd = GremlinDraw(g=g, title=title)
    gd.draw(gt)
    return gd.gviz

GremlinDrawConfig dataclass

draw configuration parameters

Source code in gremlin/draw.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@dataclass
class GremlinDrawConfig:
    """
    draw configuration parameters
    """

    fontname: str = "arial"
    fillcolor: str = "#ADE1FE"
    output_format: str = "pdf"
    edge_line_width: int = 3
    dash_width: int = 5  # number of dashes to apply
    v_limit: int = 10  # maximum number of vertices to show
    e_limit: int = 10  # maximum number of edges to show
    # optionally set the properties to be displayed
    vertex_properties: List[str] = None  # New filter for vertex properties
    edge_properties: List[str] = None  # New filter for edge properties

examples

Example dataclass

Source code in gremlin/examples.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@dataclass
class Example:
    name: str
    url: str

    def load(
        self,
        g: GraphTraversalSource,
        volume: Volume,
        force: bool = False,
        debug: bool = False,
    ) -> None:
        """
        download graph from remote_path to local_path depending on force flag
        and load graph into g

        Args:
            g(GraphTraversalSource): the target graph (inout)
            volume:Volume
            force(bool): if True download even if local copy already exists
            debug(bool): if True show debugging information
        """
        self.download(volume.local_path, force=force, debug=debug)
        graph_xml = f"{volume.remote_path}/{self.name}.xml"
        RemoteTraversal.load(g, graph_xml)

    def download(self, path, force: bool = False, debug: bool = False) -> str:
        """
        load the graphml xml file from the given url and store it to the given file_name (prefix)

        Args:
            url(str): the url to use
            file_name(str): the name of the file to load
            force(bool): if True overwrite
            debug(bool): if True show debugging information

        Returns:
            str: the filename loaded
        """
        graph_xml = f"{path}/{self.name}.xml"
        # check whether file exists and is size 0 which indicates
        # a failed download attempt
        if os.path.exists(graph_xml):
            stats = os.stat(graph_xml)
            size = stats.st_size
            force = force or size == 0
            if debug:
                print(f"{graph_xml}(size {size}) already downloaded ...")
        if not os.path.exists(graph_xml) or force:
            if debug:
                print(f"downloading {self.url} to {graph_xml} ...")
            graph_data = urllib.request.urlopen(self.url).read().decode("utf-8")
            print(graph_data, file=open(graph_xml, "w"))
        return graph_xml

download(path, force=False, debug=False)

load the graphml xml file from the given url and store it to the given file_name (prefix)

Parameters:

Name Type Description Default
url(str)

the url to use

 required
file_name(str)

the name of the file to load

 required
force(bool)

if True overwrite

 required
debug(bool)

if True show debugging information

 required

Returns:

Name Type Description
str str

the filename loaded
Source code in gremlin/examples.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def download(self, path, force: bool = False, debug: bool = False) -> str:
    """
    load the graphml xml file from the given url and store it to the given file_name (prefix)

    Args:
        url(str): the url to use
        file_name(str): the name of the file to load
        force(bool): if True overwrite
        debug(bool): if True show debugging information

    Returns:
        str: the filename loaded
    """
    graph_xml = f"{path}/{self.name}.xml"
    # check whether file exists and is size 0 which indicates
    # a failed download attempt
    if os.path.exists(graph_xml):
        stats = os.stat(graph_xml)
        size = stats.st_size
        force = force or size == 0
        if debug:
            print(f"{graph_xml}(size {size}) already downloaded ...")
    if not os.path.exists(graph_xml) or force:
        if debug:
            print(f"downloading {self.url} to {graph_xml} ...")
        graph_data = urllib.request.urlopen(self.url).read().decode("utf-8")
        print(graph_data, file=open(graph_xml, "w"))
    return graph_xml

load(g, volume, force=False, debug=False)

download graph from remote_path to local_path depending on force flag and load graph into g

Parameters:

Name Type Description Default
g(GraphTraversalSource)

the target graph (inout)

 required
volume Volume

Volume

 required
force(bool)

if True download even if local copy already exists

 required
debug(bool)

if True show debugging information

 required
Source code in gremlin/examples.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def load(
    self,
    g: GraphTraversalSource,
    volume: Volume,
    force: bool = False,
    debug: bool = False,
) -> None:
    """
    download graph from remote_path to local_path depending on force flag
    and load graph into g

    Args:
        g(GraphTraversalSource): the target graph (inout)
        volume:Volume
        force(bool): if True download even if local copy already exists
        debug(bool): if True show debugging information
    """
    self.download(volume.local_path, force=force, debug=debug)
    graph_xml = f"{volume.remote_path}/{self.name}.xml"
    RemoteTraversal.load(g, graph_xml)

Examples

Examples

Source code in gremlin/examples.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class Examples:
    """
    Examples
    """

    def __init__(self, volume: Volume, debug: bool = False):
        """
        Constructor

        Args:
            volume:Volume
            debug(bool): if true switch on debugging

        """
        self.debug = debug
        self.volume = volume
        self.examples_by_name = {}
        for example in [
            Example(
                name="tinkerpop-modern",
                url="https://raw.githubusercontent.com/apache/tinkerpop/master/data/tinkerpop-modern.xml",
            ),
            Example(
                name="grateful-dead",
                url="https://raw.githubusercontent.com/apache/tinkerpop/master/data/grateful-dead.xml",
            ),
            Example(
                name="air-routes-small",
                url="https://raw.githubusercontent.com/krlawrence/graph/master/sample-data/air-routes-small.graphml",
            ),
            Example(
                name="air-routes-latest",
                url="https://raw.githubusercontent.com/krlawrence/graph/master/sample-data/air-routes-latest.graphml",
            ),
        ]:
            self.examples_by_name[example.name] = example

    def load_by_name(self, g: GraphTraversalSource, name: str) -> None:
        """
        load an example by name to the given graph

        Args:
            g(GraphTraversalSource): the target graph (inout)
            name(str): the name of the example

        Raises:
            Exception: if the example does not exist
        """
        if name in self.examples_by_name:
            example = self.examples_by_name[name]
            example.load(g, self.volume, debug=self.debug)
        else:
            raise Exception(f"invalid example {name}")

__init__(volume, debug=False)

Constructor

Parameters:

Name Type Description Default
volume Volume

Volume

 required
debug(bool)

if true switch on debugging

 required
Source code in gremlin/examples.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def __init__(self, volume: Volume, debug: bool = False):
    """
    Constructor

    Args:
        volume:Volume
        debug(bool): if true switch on debugging

    """
    self.debug = debug
    self.volume = volume
    self.examples_by_name = {}
    for example in [
        Example(
            name="tinkerpop-modern",
            url="https://raw.githubusercontent.com/apache/tinkerpop/master/data/tinkerpop-modern.xml",
        ),
        Example(
            name="grateful-dead",
            url="https://raw.githubusercontent.com/apache/tinkerpop/master/data/grateful-dead.xml",
        ),
        Example(
            name="air-routes-small",
            url="https://raw.githubusercontent.com/krlawrence/graph/master/sample-data/air-routes-small.graphml",
        ),
        Example(
            name="air-routes-latest",
            url="https://raw.githubusercontent.com/krlawrence/graph/master/sample-data/air-routes-latest.graphml",
        ),
    ]:
        self.examples_by_name[example.name] = example

load_by_name(g, name)

load an example by name to the given graph

Parameters:

Name Type Description Default
g(GraphTraversalSource)

the target graph (inout)

 required
name(str)

the name of the example

 required

Raises:

Type Description
Exception

if the example does not exist
Source code in gremlin/examples.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def load_by_name(self, g: GraphTraversalSource, name: str) -> None:
    """
    load an example by name to the given graph

    Args:
        g(GraphTraversalSource): the target graph (inout)
        name(str): the name of the example

    Raises:
        Exception: if the example does not exist
    """
    if name in self.examples_by_name:
        example = self.examples_by_name[name]
        example.load(g, self.volume, debug=self.debug)
    else:
        raise Exception(f"invalid example {name}")

Volume dataclass

map a local path on the client to a remote path on a server e.g. when using a Volume in a docker container

Source code in gremlin/examples.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@dataclass
class Volume:
    """
    map a local path on the client to a remote
    path on a server e.g. when using a Volume in a docker
    container
    """

    local_path: str
    remote_path: str

    def local(self, file_name: str):
        """
        return the local mapping of the given file_name

        Args:
            file_name(str): the file name to map
        Returns:
            str: the local path
        """
        path = f"{self.local_path}/{file_name}"
        return path

    def remote(self, file_name: str):
        """
        return the remote mapping of the given file_name

        Args:
            file_name(str): the file name to map
        Returns:
            str: the remote path
        """
        path = f"{self.remote_path}/{file_name}"
        return path

    @staticmethod
    def docker() -> "Volume":
        """
        get the default docker volume mapping

        Returns:
            Volume: the local_path/remote_path mapping
        """
        home = str(Path.home())
        local_path = f"{home}/.gremlin-examples"
        os.makedirs(local_path, exist_ok=True)
        remote_path = "/opt/gremlin-server/data/examples"
        volume = Volume(local_path=local_path, remote_path=remote_path)
        return volume

    @staticmethod
    def local() -> "Volume":
        """
        get the default local volume mapping

        Returns:
            Volume: the local_path/remote_path mapping
        """
        home = str(Path.home())
        local_path = f"{home}/.gremlin-examples"
        os.makedirs(local_path, exist_ok=True)
        remote_path = str(abspath(f"{dirname(abspath(__file__))}/data"))
        volume = Volume(local_path=local_path, remote_path=remote_path)
        return volume

docker() staticmethod

get the default docker volume mapping

Returns:

Name Type Description
Volume Volume

the local_path/remote_path mapping
Source code in gremlin/examples.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@staticmethod
def docker() -> "Volume":
    """
    get the default docker volume mapping

    Returns:
        Volume: the local_path/remote_path mapping
    """
    home = str(Path.home())
    local_path = f"{home}/.gremlin-examples"
    os.makedirs(local_path, exist_ok=True)
    remote_path = "/opt/gremlin-server/data/examples"
    volume = Volume(local_path=local_path, remote_path=remote_path)
    return volume

local() staticmethod

get the default local volume mapping

Returns:

Name Type Description
Volume Volume

the local_path/remote_path mapping
Source code in gremlin/examples.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@staticmethod
def local() -> "Volume":
    """
    get the default local volume mapping

    Returns:
        Volume: the local_path/remote_path mapping
    """
    home = str(Path.home())
    local_path = f"{home}/.gremlin-examples"
    os.makedirs(local_path, exist_ok=True)
    remote_path = str(abspath(f"{dirname(abspath(__file__))}/data"))
    volume = Volume(local_path=local_path, remote_path=remote_path)
    return volume

remote(file_name)

return the remote mapping of the given file_name

Parameters:

Name Type Description Default
file_name(str)

the file name to map

 required

Returns: str: the remote path

Source code in gremlin/examples.py 
35
36
37
38
39
40
41
42
43
44
45
def remote(self, file_name: str):
    """
    return the remote mapping of the given file_name

    Args:
        file_name(str): the file name to map
    Returns:
        str: the remote path
    """
    path = f"{self.remote_path}/{file_name}"
    return path

remote

Created on 2019-09-17

@author: wf

RemoteTraversal

helper class for Apache Tinkerpop Gremlin Python GLV remote access

Source code in gremlin/remote.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class RemoteTraversal:
    """
    helper class for Apache Tinkerpop Gremlin Python GLV remote access
    """

    def __init__(self, server: Server, in_jupyter: bool = False) -> None:
        """
        constructor

        """
        self.server = server
        self.in_jupyter = in_jupyter

    @staticmethod
    def fromYaml(
        serverName="server", config_path: Optional[str] = None, in_jupyter: bool = False
    ) -> "RemoteTraversal":
        """
        create a server from the given yaml file

        Args:
            serverName(str): the servername to use
            config_path(str): the path to the server configuration file
        """
        server = Server.read(serverName, config_path)
        rt = RemoteTraversal(server, in_jupyter=in_jupyter)
        return rt

    def g(self) -> GraphTraversalSource:
        """
        get the graph traversal source

        Returns:
            the graph traversal source
        """
        server = self.server
        url = f"ws://{server.host}:{server.port}/gremlin"
        # https://github.com/orientechnologies/orientdb-gremlin/issues/143
        # username="root"
        # password="rootpwd"
        if self.in_jupyter:
            self.remoteConnection = DriverRemoteConnection(
                url,
                server.alias,
                username=server.username,
                password=server.password,
                transport_factory=lambda: AiohttpTransport(call_from_event_loop=True),
            )
        else:
            self.remoteConnection = DriverRemoteConnection(
                url, server.alias, username=server.username, password=server.password
            )
        g = traversal().withRemote(self.remoteConnection)
        return g

    def close(self) -> None:
        """
        close my connection
        """
        self.remoteConnection.close()

    @staticmethod
    def load(g: GraphTraversalSource, graphmlFile) -> None:
        """
        load the given graph from the given graphmlFile
        """
        # make the local file accessible to the server
        xmlPath = os.path.abspath(graphmlFile)
        # drop the existing content of the graph
        g.V().drop().iterate()
        # read the content from the graphmlFile
        g.io(xmlPath).read().iterate()

    @staticmethod
    def clear(g: GraphTraversalSource) -> None:
        """
        clear the given graph
        """
        g.V().drop().iterate()

__init__(server, in_jupyter=False)

constructor

Source code in gremlin/remote.py 
27
28
29
30
31
32
33
def __init__(self, server: Server, in_jupyter: bool = False) -> None:
    """
    constructor

    """
    self.server = server
    self.in_jupyter = in_jupyter

clear(g) staticmethod

clear the given graph

Source code in gremlin/remote.py 
 95
 96
 97
 98
 99
100
@staticmethod
def clear(g: GraphTraversalSource) -> None:
    """
    clear the given graph
    """
    g.V().drop().iterate()

close()

close my connection

Source code in gremlin/remote.py 
77
78
79
80
81
def close(self) -> None:
    """
    close my connection
    """
    self.remoteConnection.close()

fromYaml(serverName='server', config_path=None, in_jupyter=False) staticmethod

create a server from the given yaml file

Parameters:

Name Type Description Default
serverName(str)

the servername to use

 required
config_path(str)

the path to the server configuration file

 required
Source code in gremlin/remote.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def fromYaml(
    serverName="server", config_path: Optional[str] = None, in_jupyter: bool = False
) -> "RemoteTraversal":
    """
    create a server from the given yaml file

    Args:
        serverName(str): the servername to use
        config_path(str): the path to the server configuration file
    """
    server = Server.read(serverName, config_path)
    rt = RemoteTraversal(server, in_jupyter=in_jupyter)
    return rt

g()

get the graph traversal source

Returns:

Type Description
GraphTraversalSource

the graph traversal source
Source code in gremlin/remote.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def g(self) -> GraphTraversalSource:
    """
    get the graph traversal source

    Returns:
        the graph traversal source
    """
    server = self.server
    url = f"ws://{server.host}:{server.port}/gremlin"
    # https://github.com/orientechnologies/orientdb-gremlin/issues/143
    # username="root"
    # password="rootpwd"
    if self.in_jupyter:
        self.remoteConnection = DriverRemoteConnection(
            url,
            server.alias,
            username=server.username,
            password=server.password,
            transport_factory=lambda: AiohttpTransport(call_from_event_loop=True),
        )
    else:
        self.remoteConnection = DriverRemoteConnection(
            url, server.alias, username=server.username, password=server.password
        )
    g = traversal().withRemote(self.remoteConnection)
    return g

load(g, graphmlFile) staticmethod

load the given graph from the given graphmlFile

Source code in gremlin/remote.py 
83
84
85
86
87
88
89
90
91
92
93
@staticmethod
def load(g: GraphTraversalSource, graphmlFile) -> None:
    """
    load the given graph from the given graphmlFile
    """
    # make the local file accessible to the server
    xmlPath = os.path.abspath(graphmlFile)
    # drop the existing content of the graph
    g.V().drop().iterate()
    # read the content from the graphmlFile
    g.io(xmlPath).read().iterate()

Server

Server description

Source code in gremlin/remote.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class Server:
    """
    Server description
    """

    debug = False

    # construct me with the given alias
    def __init__(
        self,
        host: str = "localhost",
        port: int = 8182,
        alias: str = "g",
        name: str = "TinkerGraph",
        username: str = "",
        password: str = "",
        debug: bool = False,
        helpUrl: str = "http://wiki.bitplan.com/index.php/Gremlin_python#Connecting_to_Gremlin_enabled_graph_databases",
    ) -> None:
        """
        constructor

        Args:
            host(str): the host to connect to
            port(int): the port to connect to
            alias(str): the alias to use
            name(str): the name of the server
            username(Optional[str]): the username to use
            password(Optional[str]): the password to use
            debug(bool): True if debug output should be generated
            helpUrl(str): the help url to use
        """
        self.host = host
        self.port = port
        self.alias = alias
        self.name = name
        self.username = username
        self.password = password
        Server.debug = debug
        self.helpUrl = helpUrl

    def check_socket(self) -> bool:
        """
        check my socket

        Returns:
            True if socket is open
        """
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
            is_open = sock.connect_ex((self.host, self.port)) == 0
            return is_open

    # return a readable representation of me
    def __repr__(self) -> str:
        return "%s(%r)" % (self.__class__, self.__dict__)

    @staticmethod
    def read(name: str, config_path: Optional[str] = None) -> "Server":
        """
        read me from a yaml file

        Args:
            name(str): the name of the server
            config_path(str): the path to the config files

        Returns:
            the server

        Raises:
            Exception: if the yaml file is missing
        """
        if config_path is None:
            script_path = dirname(abspath(__file__))
            config_path = abspath(f"{script_path}/config")
        yamlFile = f"{config_path}/{name}.yaml"
        # is there a yamlFile for the given name
        if os.path.isfile(yamlFile):
            with io.open(yamlFile, "r") as stream:
                if Server.debug:
                    print("reading %s" % (yamlFile))
                server = yaml.load(stream, Loader=yaml.Loader)
                if Server.debug:
                    print(server)
                return server
        else:
            raise Exception(f"{yamlFile} is missing")

    # write me to my yaml file
    def write(self) -> None:
        """
        write me to my yaml file
        """
        yamlFile = self.name + ".yaml"
        with io.open(yamlFile, "w", encoding="utf-8") as stream:
            yaml.dump(self, stream)
            if Server.debug:
                print(yaml.dump(self))

__init__(host='localhost', port=8182, alias='g', name='TinkerGraph', username='', password='', debug=False, helpUrl='http://wiki.bitplan.com/index.php/Gremlin_python#Connecting_to_Gremlin_enabled_graph_databases')

constructor

Parameters:

Name Type Description Default
host(str)

the host to connect to

 required
port(int)

the port to connect to

 required
alias(str)

the alias to use

 required
name(str)

the name of the server

 required
username(Optional[str])

the username to use

 required
password(Optional[str])

the password to use

 required
debug(bool)

True if debug output should be generated

 required
helpUrl(str)

the help url to use

 required
Source code in gremlin/remote.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def __init__(
    self,
    host: str = "localhost",
    port: int = 8182,
    alias: str = "g",
    name: str = "TinkerGraph",
    username: str = "",
    password: str = "",
    debug: bool = False,
    helpUrl: str = "http://wiki.bitplan.com/index.php/Gremlin_python#Connecting_to_Gremlin_enabled_graph_databases",
) -> None:
    """
    constructor

    Args:
        host(str): the host to connect to
        port(int): the port to connect to
        alias(str): the alias to use
        name(str): the name of the server
        username(Optional[str]): the username to use
        password(Optional[str]): the password to use
        debug(bool): True if debug output should be generated
        helpUrl(str): the help url to use
    """
    self.host = host
    self.port = port
    self.alias = alias
    self.name = name
    self.username = username
    self.password = password
    Server.debug = debug
    self.helpUrl = helpUrl

check_socket()

check my socket

Returns:

Type Description
bool

True if socket is open
Source code in gremlin/remote.py 
144
145
146
147
148
149
150
151
152
153
def check_socket(self) -> bool:
    """
    check my socket

    Returns:
        True if socket is open
    """
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        is_open = sock.connect_ex((self.host, self.port)) == 0
        return is_open

read(name, config_path=None) staticmethod

read me from a yaml file

Parameters:

Name Type Description Default
name(str)

the name of the server

 required
config_path(str)

the path to the config files

 required

Returns:

Type Description
'Server'

the server

Raises:

Type Description
Exception

if the yaml file is missing
Source code in gremlin/remote.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@staticmethod
def read(name: str, config_path: Optional[str] = None) -> "Server":
    """
    read me from a yaml file

    Args:
        name(str): the name of the server
        config_path(str): the path to the config files

    Returns:
        the server

    Raises:
        Exception: if the yaml file is missing
    """
    if config_path is None:
        script_path = dirname(abspath(__file__))
        config_path = abspath(f"{script_path}/config")
    yamlFile = f"{config_path}/{name}.yaml"
    # is there a yamlFile for the given name
    if os.path.isfile(yamlFile):
        with io.open(yamlFile, "r") as stream:
            if Server.debug:
                print("reading %s" % (yamlFile))
            server = yaml.load(stream, Loader=yaml.Loader)
            if Server.debug:
                print(server)
            return server
    else:
        raise Exception(f"{yamlFile} is missing")

write()

write me to my yaml file

Source code in gremlin/remote.py 
191
192
193
194
195
196
197
198
199
def write(self) -> None:
    """
    write me to my yaml file
    """
    yamlFile = self.name + ".yaml"
    with io.open(yamlFile, "w", encoding="utf-8") as stream:
        yaml.dump(self, stream)
        if Server.debug:
            print(yaml.dump(self))