Skip to content

Instantly share code, notes, and snippets.

@enjoylife
Created November 3, 2011 21:42
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save enjoylife/1337872 to your computer and use it in GitHub Desktop.
Save enjoylife/1337872 to your computer and use it in GitHub Desktop.
testing of Flask and Neo4j
import os
import jpype
from neo4j import GraphDatabase
from flask import Flask, request, session, g, redirect, url_for, abort, render_template, flash
app = Flask(__name__)
app.config.from_object(__name__)
app.config['ADMIN'] ='Matthew'
app.config['DATABASE']='/tmp/graphtest'
def connectDB():
return GraphDatabase(app.config['DATABASE'])
def initDB():
db = connectDB()
with db.transaction:
users = db.node()
roles = db.node()
db.reference_node.USERS(users)
db.reference_node.ROLES(roles)
userIndex = db.node.indexes.create(users)
user = db.node(name=app.config['ADMIN'])
user.INSTANCE_OF(users)
userIndex['name'][app.config['ADMIN']] = user
role = db.node(type='superadmin')
role.INSTANCE_OF(roles)
role.ASSIGN_TO(user)
db.shutdown()
print "Database initialized."
def testDB():
db = connectDB()
with db.transaction:
userIndex = db.node.indexes.get('users')
user = userIndex['name'][app.config['ADMIN']].single
username = user['name']
db.shutdown()
print "Admin username is '%s'. Database exists." % username
@app.before_request
def before_request():
jpype.attachThreadToJVM()
g.db = connectDB()
@app.teardown_request
def teardown_request(exception):
g.db.shutdown()
@app.route('/')
def index():
with g.db.transaction:
userIndex = g.db.node.indexes.get('users')
user = userIndex['name'][app.config['ADMIN']].single
username = user['name']
fields = dict(username=username)
return str(fields)
# return render_template('index.html', fields=fields)
if os.path.exists(app.config['DATABASE']) == False:
initDB()
else:
testDB()
if __name__ == "__main__":
app.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment