This example project demos integration between Graphene, Sanic and SQLAlchemy.
The project contains two models, one named Department
, Role
and another
named Employee
.
Sanic example using Graphsql + SQLAlchemy
Run app.py and access via localhost:5000/graphql
- Graphene
- Sanic
- Graphene-Sqlalchemy
- Sqlalchemy
#Tutzz
- Build your model first
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import backref, relationships
from database import Base
# Basic SQLAlchemy
class Department(Base):
__tablename__ = 'department'
id = Column(Integer, primary_key=True)
name = Column(String)
class Role(Base):
__tablename__ = 'roles'
role_id = Column(Integer, primary_key=True)
name = Column(String)
class Employee(Base):
__tablename__ = 'employees'
employee_id = Column(Integer, primary_key=True)
name = Column(String)
hired_on = Column(DateTime, default=func.now())
department_id = Column(Integer, ForeignKey('department.id'))
role_id = Column(Integer, ForeignKey('roles.role_id'))
department = relationships(
Department,
backref=backref('employee',
uselist=True,
cascade='delete,all'))
role = relationships(
Role,
backref=backref('roles',
uselist=True,
cascade='delete,all'))
- Initiate database + seed data + Commit
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
# Create database
engine = create_engine('sqlite:///database.sqlite3', convert_unicode=True)
db_session = scoped_session(sessionmaker(autocommit=False,
autoflush=False,
bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()
async def init_db():
from models import Department, Employee, Role
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
# Create Fixture for department
engineering = Department(name='Engineering')
db_session.add(engineering)
hr = Department(name='Human Resource')
db_session.add(hr)
# Create Fixture for Role
manager = Role(name='manager')
db_session.add(manager)
engineer = Role(name='Engineer')
db_session.add(engineer)
# Create
peter = Employee(name='Peter', department=engineering, role=manager)
db_session.add(peter)
roy = Employee(name='Roy', department=engineering, role=engineer)
db_session.add(roy)
tracy = Employee(name='Tracy', department=hr, role=manager)
db_session.add(tracy)
# Insert seed data into database
db_session.commit()
- Define your scheme for graphql It is important to know some important terms here like SQLAlchemyConnectionField and relay.Node. Check out graphene official documentation for more info
import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType
from models import Department as DepartmentModel
from models import Employee as EmployeeModel
from models import Role as RoleModel
class Department(SQLAlchemyObjectType):
class Meta:
model = DepartmentModel
interfaces = (relay.Node, )
class Employee(SQLAlchemyObjectType):
class Meta:
model = EmployeeModel
interfaces = (relay.Node, )
class Role(SQLAlchemyObjectType):
class Meta:
model = RoleModel
interfaces = (relay.Node, )
class Query(graphene.ObjectType):
node = relay.Node.Field()
all_employee = SQLAlchemyConnectionField(Employee)
all_roles = SQLAlchemyConnectionField(Role)
role = graphene.Field(Role)
schema = graphene.Schema(query=Query, types=[Department, Employee, Role])
- Let's run our Sanic App
from sanic_graphql import GraphQLView
from sanic import Sanic
from database import db_session, init_db
from schema import schema
app = Sanic(__name__)
app.debug = True
# This will be the query
default_query = """
{
allEmployee {
edges {
node {
id,
name,
department {
id,
name
},
role {
id,
name
}
}
}
}
}
""".strip()
# It's like HTTP routes but instead, it directs you to /graphql (Schema).
# graphiq = True will initiate the GUI
app.add_route(GraphQLView.as_view(schema=schema, graphiql=True), '/graphql')
# runs the app and database. init_db contains seed data.
if __name__ == '__main__':
init_db()
app.run()
EOF Fully focus on my own design agency now.