Nebula Carina is a Pydantic based ORM framework for graph database Nebula Graph.
This project is designed to provide an easy-to-use orm package for the Nebula Graph database for Python development, especially web application development.
Nebula Graph is a powerful graph database, the only graph database solution in the world that can accommodate hundreds of billions of vertices and trillions of edges and provide millisecond-level query latency.
The goals of this project include providing an object-oriented description of the database structure, concise Query statements, and JSONizable results. In particular, I used the pydantic library with the expectation that this library can be used in conjunction with any modern python web framework.
This project is based on the official package nebula-python https://github.com/vesoft-inc/nebula-python.
At present, the project is not stable yet, the code changes rapidly and the methods are unstable. If you have any ideas for improving this project, you are very welcome to contact me via issue or email.
- Python (3.10)
- nebula3-python (>=3.10)
install using pip
pip install nebula-carina
If you are using Django framework, then add nebula_carina
to INSTALLED_APPS
:
INSTALLED_APPS = [
...
'nebula_carina',
...
]
Then, setup the CARINA_SETTINGS
in your settings.py file as following:
CARINA_SETTINGS = {
"auto_create_default_space_with_vid_desc": "FIXED_STRING(20)",
"default_space": "main",
"max_connection_pool_size": 10,
"model_paths": ["nebula.carina"],
"user_name": "root",
"password": "1234",
"servers": ["192.168.31.248:9669"],
"timezone_name": "UTC",
}
Please configure environment variables if you are not using Django.
nebula_servers='["192.168.1.10:9669"]'
nebula_user_name=root
nebula_password=1234
nebula_max_connection_pool_size=10
nebula_model_paths='["example.models"]'
nebula_default_space=main
nebula_auto_create_default_space_with_vid_desc=FIXED_STRING(20)
nebula_timezone_name=UTC
You can export by:
export nebula_model_paths='["example.models"]' nebula_password=1234 nebula_servers='["192.168.1.10:9669"]' nebula_user_name=root nebula_default_space=main nebula_auto_create_default_space_with_vid_desc=FIXED_STRING(20)
Ensure that the default space exists. You can create a default space by creating a script:
from nebula_carina.ngql.schema.space import create_space, show_spaces, VidTypeEnum
main_space_name = "main"
if main_space_name not in show_spaces():
create_space(main_space_name, (VidTypeEnum.FIXED_STRING, 20))
Or you can just set nebula_auto_create_default_space_with_vid_desc=FIXED_STRING(20)
so that it will be automatically handled.
Then, develop models defined in nebula_model_paths.
- An TagModel is used to define a nebula tag.
- An EdgeTypeModel is used to define a nebula edge type.
For example:
from datetime import datetime
from nebula_carina.models import models
from nebula_carina.models.fields import create_nebula_field as _
from nebula_carina.ngql.schema import data_types
class Figure(models.TagModel):
name: str = _(data_types.FixedString(30), ..., )
age: int = _(data_types.Int16, ..., )
valid_until: int = _(data_types.Int64, None, )
hp: int = _(data_types.Int16, 100, )
style: str = _(data_types.FixedString(10), 'rap', )
is_virtual: bool = _(data_types.Bool, True)
created_on: datetime = _(data_types.Datetime, data_types.Datetime.auto)
some_dt: datetime = _(data_types.Datetime, datetime(2022, 1, 1))
class Meta:
ttl_duration = 100
ttl_col = 'valid_until'
class Source(models.TagModel):
name: str = _(data_types.FixedString(30), ..., )
class Belong(models.EdgeTypeModel):
pass
class Love(models.EdgeTypeModel):
way: str = _(data_types.FixedString(10), ..., )
times: int = _(data_types.Int8, ..., )
class Support(models.EdgeTypeModel):
food_amount: int = _(data_types.Int16, ..., )
- An VertexModel is used to define a nebula vertex. It does nothing to the schema.
from nebula_carina.models import models
class VirtualCharacter(models.VertexModel):
figure: Figure
source: Source
class LimitedCharacter(models.VertexModel):
figure: Figure
- An EdgeModel is used to define a nebula edge. But note that there will be no subclasses for edge model since we don't need it.
Use make_migrations
and migrate
to synchronize the schema to current space.
Please note that only final consistency on DB schema is currently supported.
from nebula_carina.models.migrations import make_migrations, migrate
make_migrations()
# you can print out the result and check it
# then, run migrate
migrate(make_migrations())
If you are using Django, you can do the migration by the following command:
python manage.py nebulamigrate
from example.models import VirtualCharacter, Figure, Source, LimitedCharacter, Love, Support
from nebula_carina.models.models import EdgeModel, Limit
from datetime import datetime
# create/update vertex
VirtualCharacter(
vid='char_test1', figure=Figure(
name='test1', age=100, is_virtual=False, some_dt=datetime(2021, 3, 3, 0, 0, 0, 12)
), source=Source(name='movie1')
).save()
VirtualCharacter(
vid='char_test2', figure=Figure(
name='test2', age=30, is_virtual=False, some_dt=datetime(2021, 3, 3, 0, 0, 0, 12)
), source=Source(name='movie2')
).save()
LimitedCharacter(
vid='char_test3', figure=Figure(
name='test3', age=200, is_virtual=True, some_dt=datetime(2022, 3, 3, 0, 0, 0, 12)
)
).save()
# note that a VirtualCharacter could still be recognized as a LimitedCharacter, but a LimitedCharacter is not a VirtualCharacter
# create/update edge
EdgeModel(src_vid='char_test1', dst_vid='char_test2', ranking=0, edge_type=Love(way='gun', times=40)).save()
# create/update another edge
EdgeModel(src_vid='char_test1', dst_vid='char_test2', ranking=0, edge_type=Support(food_amount=400)).save()
# get vertex by id
character1 = VirtualCharacter.objects.get('char_test1')
character2 = VirtualCharacter.objects.get('char_test2')
# find the exact edge by src_vid, dst_vid and edge type
edge1 = EdgeModel.objects.get('char_test1', 'char_test2', Love)
# find all edges between vertexes
EdgeModel.objects.find_between('char_test1', 'char_test2', limit=Limit(10))
# find specific edges between vertexes
EdgeModel.objects.find_between('char_test1', 'char_test2', Support)
# find vertexes (sources) that go towards node by the specific edge type
VirtualCharacter.objects.find_sources('char_test2', Love, distinct=True, limit=Limit(1))
# Or just by any edge
VirtualCharacter.objects.find_sources('char_test2')
# or just
character2.get_sources(Love, VirtualCharacter)
# similarly, find the destinations
VirtualCharacter.objects.find_destinations('char_test1', Love)
# or
character1.get_destinations(Love, VirtualCharacter)
# out edges & reverse edges
character1.get_out_edges(Love)
character2.get_reverse_edges(Love)
# get edges & vertexes together
character1.get_out_edge_and_destinations(Love, VirtualCharacter)
character2.get_reverse_edge_and_sources(Love, VirtualCharacter)
- A easy model builder ready for you to build any ngql
from nebula_carina.ngql.query.conditions import Q
from nebula_carina.models.model_builder import ModelBuilder
from nebula_carina.ngql.query.match import Limit
from nebula_carina.models.models import EdgeModel
ModelBuilder.match('(v:figure:source)', {'v': VirtualCharacter}, limit=Limit(10))
ModelBuilder.match(
'(v)-[e:love]->(v2)', {'v': VirtualCharacter, 'e': EdgeModel, 'v2': VirtualCharacter},
condition=Q(v__id__in=["char_test1", "char_test3"]),
)
If you are using fastapi, then serialization and deserialization are already handled by the repo. For example, in your api functions, you are welcomed to use the result of data model or the model builder in your return function. It's very easy to use!
from fastapi import FastAPI
from example.models import VirtualCharacter, Love
from nebula_carina.models.model_builder import ModelBuilder
from nebula_carina.models.models import EdgeModel
from nebula_carina.ngql.query.conditions import Q
app = FastAPI()
@app.get("/character/{character_id}")
async def get_character(character_id: str):
return VirtualCharacter.objects.get(character_id)
@app.get("/character/{character_id}/admirers")
async def get_admirers(character_id: str):
return VirtualCharacter.objects.find_sources(character_id, Love, distinct=True)
@app.get("/character/{character_id}/your-complex-relation")
async def what_a_complex_human_relation(character_id: str):
return ModelBuilder.match(
'(v)-[e:love]->(v2)-[e2:love]->(v3)', {
'v': VirtualCharacter, 'e': EdgeModel, 'v2': VirtualCharacter,
'e2': EdgeModel, 'v3': VirtualCharacter
},
condition=Q(v__id=character_id),
)
Basically things are the same as in fastapi except that you have to use .dict()
method to serialize a model before using it in response.
In django, you can use ModelBuilder.serialized_match
method to directly get the serialized result of match method.
from example.models import VirtualCharacter
from django.http import JsonResponse
from nebula_carina.models.models import EdgeModel
from nebula_carina.ngql.query.conditions import Q
from nebula_carina.models.model_builder import ModelBuilder
def some_view(request, character_id: str):
vr = VirtualCharacter.objects.get(character_id)
# make sure that use .dict() function to serialize the result
return JsonResponse(vr.dict())
def what_a_complex_human_relation(request, character_id: str):
return JsonResponse(ModelBuilder.serialized_match(
'(v)-[e:love]->(v2)-[e2:love]->(v3)', {
'v': VirtualCharacter, 'e': EdgeModel, 'v2': VirtualCharacter,
'e2': EdgeModel, 'v3': VirtualCharacter
},
condition=Q(v__id=character_id),
), safe=False) # returns a list of dict with keys {v, e, v2, e2, v3}
# same method as the previous one if you would like to access the models
def what_a_complex_human_relation_another(request, character_id: str):
return JsonResponse([
single_match_result.dict() for single_match_result in ModelBuilder.match(
'(v)-[e:love]->(v2)-[e2:love]->(v3)', {
'v': VirtualCharacter, 'e': EdgeModel, 'v2': VirtualCharacter,
'e2': EdgeModel, 'v3': VirtualCharacter
},
condition=Q(v__id=character_id),
)
], safe=False) # returns a list of dict with keys {v, e, v2, e2, v3}
Flask usage is quite similar to the Django usage. Basically use .dict()
function to serialize the model.
- Indexes
- TTL on schema
- Go / Fetch / Lookup statements
- Session / Connection Pool (partially, might implement a better one later base on nebula-python)
- More abstractions on different scenarios
- Default values for schema models
- Generic Vertex Model
- Basic Django Support
- Django management.py
- Match Result Wrapper