Integrate Graphene with Sqlalchemy and Flask

Overview

这篇文章介绍了如何用 graphene + flask + flask_graphql + sqlalchemy + graphene_sqlalchemy 实现一个极简但跟生产环境的代码很类似的的 GraphQL API. 这个例子中我们用 sqlite 做数据库

Note

这个例子来自于 https://docs.graphene-python.org/projects/sqlalchemy/en/latest/tutorial/

Code

models 模块定义了 sqlalchemy ORM 的模型.

models.py

# -*- coding: utf-8 -*-

import sqlalchemy as sa
import sqlalchemy.orm as orm

engine = sa.create_engine("sqlite:///database.sqlite3", convert_unicode=True)
db_session = orm.scoped_session(
    orm.sessionmaker(autocommit=False, autoflush=False, bind=engine)
)

Base = orm.declarative_base()
# We will need this for querying
Base.query = db_session.query_property()


class Department(Base):
    __tablename__ = "department"

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)


class Employee(Base):
    __tablename__ = "employee"

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)
    hired_on = sa.Column(sa.DateTime, default=sa.func.now())
    department_id = sa.Column(sa.Integer, sa.ForeignKey("department.id"))

    department = orm.relationship(
        Department,
        backref=orm.backref("employees", uselist=True, cascade="delete,all"),
    )

运行 add_some_data.py 脚本可以往数据库中添加一些测试用的数据.

models.py

# -*- coding: utf-8 -*-

import sqlalchemy as sa
import sqlalchemy.orm as orm

engine = sa.create_engine("sqlite:///database.sqlite3", convert_unicode=True)
db_session = orm.scoped_session(
    orm.sessionmaker(autocommit=False, autoflush=False, bind=engine)
)

Base = orm.declarative_base()
# We will need this for querying
Base.query = db_session.query_property()


class Department(Base):
    __tablename__ = "department"

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)


class Employee(Base):
    __tablename__ = "employee"

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)
    hired_on = sa.Column(sa.DateTime, default=sa.func.now())
    department_id = sa.Column(sa.Integer, sa.ForeignKey("department.id"))

    department = orm.relationship(
        Department,
        backref=orm.backref("employees", uselist=True, cascade="delete,all"),
    )

schema 模块定义了 GraphQL 的 schema. graphene_sqlalchemy 库能够自动将 sqlalchemy ORM 对象转化为 graphql 对象, 并且自动实现了对应的 resolver, 而无需你手动实现.

schema.py

# -*- coding: utf-8 -*-

import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from models import db_session, Department as DepartmentModel, Employee as EmployeeModel


class Department(SQLAlchemyObjectType):
    class Meta:
        model = DepartmentModel
        interfaces = (relay.Node,)


class Employee(SQLAlchemyObjectType):
    class Meta:
        model = EmployeeModel
        interfaces = (relay.Node,)


class Query(graphene.ObjectType):
    node = relay.Node.Field()
    # Allows sorting over multiple columns, by default over the primary key
    all_employees = SQLAlchemyConnectionField(Employee.connection)
    # Disable sorting over this field
    all_departments = SQLAlchemyConnectionField(Department.connection, sort=None)


schema = graphene.Schema(query=Query)

app 模块将 graphql 的部分和 flask 的部分结合在一起.

app.py

# -*- coding: utf-8 -*-

import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from models import db_session, Department as DepartmentModel, Employee as EmployeeModel


class Department(SQLAlchemyObjectType):
    class Meta:
        model = DepartmentModel
        interfaces = (relay.Node,)


class Employee(SQLAlchemyObjectType):
    class Meta:
        model = EmployeeModel
        interfaces = (relay.Node,)


class Query(graphene.ObjectType):
    node = relay.Node.Field()
    # Allows sorting over multiple columns, by default over the primary key
    all_employees = SQLAlchemyConnectionField(Employee.connection)
    # Disable sorting over this field
    all_departments = SQLAlchemyConnectionField(Department.connection, sort=None)


schema = graphene.Schema(query=Query)

Test

我们写了个简单的脚本, 用于测试在 localhost 运行的 GraphQL API Server.

test.py

# -*- coding: utf-8 -*-

import requests
from rich import print as rprint


def run_query(query: str) -> dict:
    res = requests.post(
        "http://127.0.0.1:5000/graphql",
        headers={
            "Content-Type": "application/json",
        },
        json={"query": query.strip()},
    )
    return res.json()


query = """
{
  allEmployees {
    edges {
      node {
        id
        name
        department {
          name
        }
      }
    }
  }
}
""".strip()
rprint(run_query(query))
"""
This will print:

{
    'data': {
        'allEmployees': {
            'edges': [
                {
                    'node': {
                        'id': 'RW1wbG95ZWU6MQ==',
                        'name': 'Peter',
                        'department': {'name': 'Engineering'}
                    }
                },
                {
                    'node': {
                        'id': 'RW1wbG95ZWU6Mg==',
                        'name': 'Roy',
                        'department': {'name': 'Engineering'}
                    }
                },
                {
                    'node': {
                        'id': 'RW1wbG95ZWU6Mw==',
                        'name': 'Tracy',
                        'department': {'name': 'Human Resources'}
                    }
                }
            ]
        }
    }
}
"""

Reference

• https://docs.graphene-python.org/projects/sqlalchemy/en/latest/tutorial/