Main target
In this post we will present a simple demo using flask , a python server to access a graph database neo4j with automatical doc generator swagger and create a accessable swagger UI.
Dependencies
Install flask, and swagger in local machine, and be sure you have a neo4j remote or local server available. Generally you need url, name and password to acccess neo4j and create a driver.
Be sure that these dependencies works fine:
from flask import Flask, g, request, send_from_directory, abort, request_started
from flask_cors import CORS
from flask_restful import Resource, reqparse
from flask_restful_swagger_2 import Api, swagger, Schema
from flask_json import FlaskJSON, json_response
from neo4j import GraphDatabase, basic_auth
from neo4j.exceptions import Neo4jError
import neo4j.time
from dotenv import load_dotenv, find_dotenv
from datetime import datetime
import sys
import os
Step 1. Create data structure
Assuming that you have the main flask file app.py within a folder called /api, which is the name of your flask module. Create a sub-folder /api/modules/ to save your entity.
For example, if we have a data structure genre, create such file in /api/modules/genre.py/
#/api/modules/genre.py
from flask_restful_swagger_2 import Schema
class GenreModel(Schema):
type = 'object'
properties = {
'id': {
'type': 'integer',
},
'name': {
'type': 'string',
}
}
def serialize_genre(genre):
return {
'id': genre['id'],
'name': genre['name'],
}
#Notes: You can modify serialize function
And create a __init__.py in modules folder.
#api/modules/__init__.py
from .genre import *
Step 2 Establish database driver
Now go back to our app.py in api.
First we should create a .env file next to app.py to save database information, or you can save them in your system environment variables.
#api/.env
DATABASE_USERNAME="neo4j"
DATABASE_PASSWORD="your-database-password"
DATABASE_URL="bolt://localhost:7687"
FLASK_ENV=development
Then, the codes below will start a flask app, config swagger and config the driver:
load_dotenv(find_dotenv())
app = Flask(__name__)
CORS(app)
FlaskJSON(app)
api = Api(app, title='Your app name', api_version='0.1.10')
def env(key, default=None, required=True):
try:
value = os.environ[key]
return ast.literal_eval(value)
except (SyntaxError, ValueError):
return value
except KeyError:
if default or not required:
return default
raise RuntimeError("Missing required environment variable '%s'" % key)
@api.representation('application/json')
def output_json(data, code, headers=None):
return json_response(data_=data, headers_=headers, status_=code)
DATABASE_USERNAME = env('DATABASE_USERNAME')
DATABASE_PASSWORD = env('DATABASE_PASSWORD')
DATABASE_URL = env('DATABASE_URL')
driver = GraphDatabase.driver(DATABASE_URL, auth=basic_auth(DATABASE_USERNAME, str(DATABASE_PASSWORD)))
def get_db():
if not hasattr(g, 'neo4j_db'):
g.neo4j_db = driver.session()
return g.neo4j_db
@app.teardown_appcontext
def close_db(error):
if hasattr(g, 'neo4j_db'):
g.neo4j_db.close()
With the code above, we now have the important function get_db(), while the close procedure will be execute automatically by flask context. If you do not understand g and context here, please check flask's doc.
Step 3 Build the first API with swagger
Now we can write the first Restful-API under swagger. Here we define a basic API to create/get/delete a genre nodes in neo4j that we defined above.
The basic idea is :
- When calling url
api/v0/genrewith POST and a name and id in request body , create a genre. - When calling same url with GET and a id in path, return name and id.
First , import genre.py in app.py :
from api.modules import *
Then, create a class under the annotation of swagger, such class will override functions like post(), delete(),get()
class CreateGenre(Resource):
@swagger.doc({
'tags': ['genres'],
'summary': 'create genre',
'description': 'Create a genre',
'parameters': [
{
'name': 'body',
'in': 'body',
'schema': {
'type': 'object',
'properties': {
'id': {
'type': 'string',
},
'name': {
'type': 'string',
}
}
}
},
],
'responses': {
'200': {
'description': 'Genre created',
'schema': GenreModel,
}
}
})
def post(self):
data = request.get_json()
name = data.get('name')
id = data.get('id')
def _cypher_create_genre(tx,id,name):
return tx.run(
'''
MERGE (n:Genre {id: $id, name: $name }) RETURN n
''', {'id': id,'name':name}
)
db = get_db()
result = db.write_transaction(_cypher_create_genre,id,name)
return [serialize_genre(record['n']) for record in result][0]
class GetGenre(Resource):
@swagger.doc({
'tags': ['genres'],
'summary': 'get genre or delete genre',
'description': 'Genre GET/DELETE',
'parameters': [
{
'name': 'id',
'description': 'genre id',
'in': 'path',
'type': 'string',
'required': True
}
],
'responses': {
'200': {
'description': 'A genre',
'schema': GenreModel,
},
'404':{
'description': 'Genre not found'
}
}
})
def get(self,id):
def _cypher_get_genre(tx,id):
return tx.run(
'''
MATCH (n:Genre {id: $id) RETURN n
''', {'id': id}
)
db = get_db()
result = db.read_transaction(_cypher_get_genre,id)
return [serialize_genre(record['n']) for record in result][0]
Since we just declared these two classes, now we just need to map them into target URLs:
api.add_resource(CreateGenre, '/api/v1/Genre/create')
api.add_resource(CeteGenre, '/api/v1/Genre/<string:id>')
Now, when running flask, you will get a default swagger page showing these APIs.