Cog
Embedded Graph Database for Python. Lives inside your Python process. Quick setup. No server. Runs in notebooks, apps, even your browser.
Install / Use
/learn @arun1729/CogREADME
Quick Start
pip install cogdb
from cog.torque import Graph
# Create a graph and add data
g = Graph("social")
g.put("alice", "follows", "bob")
g.put("bob", "follows", "charlie")
g.put("bob", "status", "active")
# Query
g.v("alice").out("follows").all() # → {'result': [{'id': 'bob'}]}
g.v().has("status", "active").all() # → {'result': [{'id': 'bob'}]}
g.v("alice").out("follows").out("follows").all() # → {'result': [{'id': 'charlie'}]}
# Serve your graph over HTTP
g.serve() # Now queryable at http://localhost:8080
# Expose to the internet with ngrok
# $ ngrok http 8080
# Query your graph from anywhere: https://your-ngrok-url.ngrok.io
Full documentation at cogdb.io
Overview
CogDB is a persistent, embedded graph database implemented in Python. It comes with its own graph traversal language called Torque. Torque is a fluent, chainable API that lives in your Python code. There are no query strings and no context switching, you write graph queries the same way you write normal Python.
CogDB is designed to be easy to use and quick to integrate. There is no server to run and no setup overhead. You simply import it into your Python application. CogDB also works well in interactive environments such as IPython and Jupyter notebooks.
CogDB is a triple store. It models data as Node → Edge → Node (eg: Alice → Follows → Bob).
Key-value pairs let you store facts; triples let you store relationships. With a source, a destination, and a label, you get the full expressive power of a graph with only one step more structure than a key-value store.
Documentation
Creating a graph
Using put to insert triples
from cog.torque import Graph
g = Graph("people")
g.put("alice","follows","bob")
g.put("bob","follows","fred")
g.put("bob","status","cool_person")
g.put("charlie","follows","bob")
g.put("charlie","follows","dani")
g.put("dani","follows","bob")
g.put("dani","follows","greg")
g.put("dani","status","cool_person")
g.put("emily","follows","fred")
g.put("fred","follows","greg")
g.put("greg","status","cool_person")
g.put("bob","score","5")
g.put("greg","score","10")
g.put("alice","score","7")
g.put("dani","score","100")
Drop Edge
g.drop("bob", "follows", "fred")
Using putj to insert JSONs
f = Graph("followers")
f.putj('{"name" : "bob", "status" : "cool_person", "follows" : ["fred", "dani"]}')
f.putj('{"_id": "1", "name" : "fred", "status" : "cool_person", "follows" : ["alice", "greg"]}')
Using updatej to update JSONs
g.updatej('{"_id" : "1", "status" : "not_cool"}')
Torque query examples
Scan vertices
g.scan(3)
{'result': [{'id': 'bob'}, {'id': 'emily'}, {'id': 'charlie'}]}
Scan edges
g.scan(3, 'e')
{'result': [{'id': 'status'}, {'id': 'follows'}]}
Starting from a vertex, follow all outgoing edges and list all vertices
g.v("bob").out().all()
{'result': [{'id': '5'}, {'id': 'fred'}, {'id': 'cool_person'}]}
Everyone with status 'cool_person'
g.v().has("status", 'cool_person').all()
{'result': [{'id': 'bob'}, {'id': 'dani'}, {'id': 'greg'}]}
Include edges in the results
g.v().has("follows", "fred").inc().all('e')
{'result': [{'id': 'dani', 'edges': ['follows']}, {'id': 'charlie', 'edges': ['follows']}, {'id': 'alice', 'edges': ['follows']}]}
starting from a vertex, follow all outgoing edges and count vertices
g.v("bob").out().count()
'3'
See who is following who and create a view of that network
Note: render() is supported only in IPython environment like Jupyter notebook otherwise use view(..).url.
By tagging the vertices 'from' and 'to', the resulting graph can be visualized.
g.v().tag("from").out("follows").tag("to").view("follows").render()
g.v().tag("from").out("follows").tag("to").view("follows").url
file:///Path/to/your/cog_home/views/follows.html
List all views
g.lsv()
['follows']
Load existing visualization
g.getv('follows').render()
starting from a vertex, follow all out going edges and tag them
g.v("bob").out().tag("from").out().tag("to").all()
{'result': [{'from': 'fred', 'id': 'greg', 'to': 'greg'}]}
starting from a vertex, follow all incoming edges and list all vertices
g.v("bob").inc().all()
{'result': [{'id': 'alice'}, {'id': 'charlie'}, {'id': 'dani'}]}
Filtering
g.v().filter(func=lambda x: x.startswith("d")).all()
{'result': [{'id': 'dani'}]}
g.v().out("score").filter(func=lambda x: int(x) > 5).inc().all()
{'result': [{'id': 'alice'}, {'id': 'dani'}, {'id': 'greg'}]}
g.v("emily").out("follows").filter(func=lambda x: x.startswith("f")).all()
{'result': [{'id': 'fred'}]}
Bidirectional Traversal
Follow edges in both directions (outgoing and incoming):
g.v("bob").both("follows").all()
{'result': [{'id': 'fred'}, {'id': 'alice'}, {'id': 'charlie'}, {'id': 'dani'}]}
Filter to Specific Nodes
Filter results to only include specific vertices:
g.v("alice").out("follows").is_("bob", "dani").all()
{'result': [{'id': 'bob'}, {'id': 'dani'}]}
Remove Duplicates
Remove duplicate vertices from results:
g.v().out("follows").unique().all()
{'result': [{'id': 'bob'}, {'id': 'fred'}, {'id': 'greg'}, {'id': 'dani'}]}
Pagination with Limit and Skip
Limit results to first N vertices:
g.v().limit(3).all()
{'result': [{'id': 'alice'}, {'id': 'bob'}, {'id': 'charlie'}]}
Skip first N vertices:
g.v().skip(2).limit(2).all()
{'result': [{'id': 'charlie'}, {'id': 'dani'}]}
Navigate Back to Tagged Vertex
Return to a previously tagged position while preserving the traversal path:
g.v("alice").tag("start").out("follows").out("follows").back("start").all()
{'result': [{'start': 'alice', 'id': 'alice'}]}
BFS and DFS Traversal
Breadth-first search (level by level, finds shortest paths):
g.v("alice").bfs(predicates="follows", max_depth=2).all()
{'result': [{'id': 'bob'}, {'id': 'charlie'}]}
Depth-first search (explores deep before backtracking):
g.v("alice").dfs(predicates="follows", max_depth=3).all()
Filter by depth range:
# Only vertices at exactly depth 2
g.v("alice").bfs(max_depth=2, min_depth=2).all()
Stop at a target:
g.v("alice").bfs(until=lambda v: v == "fred").all()
Bidirectional traversal:
g.v("bob").bfs(direction="both", max_depth=2).all()
Using put_batch for bulk inserts (faster)
from cog.torque import Graph
g = Graph("people")
# Insert multiple triples at once - significantly faster for large graphs
g.put_batch([
("alice", "follows", "bob"),
("bob", "follows", "charlie"),
("charlie", "follows", "alice"),
("alice", "likes", "pizza"),
("bob", "likes", "tacos"),
])
Performance Tuning
Control flush behavior for faster bulk inserts:
# Default: flush every write (safest)
g = Graph("mydb")
# Fast mode: flush every 100 writes (auto-enables async)
g = Graph("mydb", flush_interval=100)
# Manual flush only (fastest for bulk loads)
g = Graph("mydb", flush_interval=0)
g.put_batch(large_dataset)
g.sync() # Flush when done
| flush_interval | Behavior | Use Case |
|------------------|----------|----------|
| 1 (default) | Flush every write | Interactive, safe |
| > 1 | Async flush every N writes | Bulk inserts |
| 0 | Manual only (sync()) | Maximum speed |
Serving a Graph Over Network
Serve a graph over HTTP and query it from another process or machine:
from cog.torque import Graph
# Create and serve a graph
g = Graph("social")
g.put("alice", "follows", "bob")
g.put("bob", "follows", "charlie")
g.serve(port=8080)
Query from another Python process:
from cog.torque import Graph
# Connect to remote graph
remote = Graph.connect("http://localhost:8080/social")
# Query just like a local graph
remote.v("alice").out("follows").all()
# {'result': [{'id': 'bob'}]}
Stop the server:
g.stop()
json example
#### Using `putj` to insert JSONs
f = Graph("followers")
f.putj('{"name" : "bob", "status" : "cool_person", "follows" : ["fred", "dani"]}')
f.putj('{"name" : "fred", "status" : "cool_person", "follows" : ["alice", "greg"]}')
f.v().has('name','bob').out('f
