Flask + GraphQL

REST APIs have a fixed shape problem: clients either get too much data or have to make multiple requests to get what they need. GraphQL flips that, the client declares exactly what it wants and the server delivers exactly that. No extra fields, no N+1 requests to stitch together related data.

TL;DR: Set up a Flask + GraphQL API with graphene and psycopg2, defining types, queries, and resolvers backed onto a PostgreSQL database.
Stack: Python, Flask, GraphQL, graphene, psycopg2, PostgreSQL
Level: Intermediate
Reading time: ~8 min

Install libs

pip install flask Flask-GraphQL graphene graphql-core graphql-relay graphql-server-core python-dotenv psycopg2

Create app.py

from flask import Flask
from flask_graphql import GraphQLView
from schemas import get_schema

app = Flask(__name__)

app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view('graphql', schema=get_schema(), graphiql=True)
)

if __name__ == '__main__':
    app.run(debug=True)

Implement schemas.py

This file defines all your GraphQL types, the Query class, and the resolver methods that fetch data from the database. Every field on a type that requires a database lookup gets its own resolve_* method. By convention, every attribute has a corresponding implementation with the prefix “resolve”.

import graphene
import psycopg2
from dotenv import load_dotenv
import os

load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")

class Position(graphene.ObjectType):
    id = graphene.Int()
    description = graphene.String()

class Professional(graphene.ObjectType):
    id = graphene.Int()
    name = graphene.String()
    position = graphene.Field(Position)
    position_id = graphene.Int()

    def resolve_position(self, info):
        with psycopg2.connect(DATABASE_URL) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT id, description FROM djangoapp_position WHERE id = %s",
                    (self.position_id,)
                )
                data = cur.fetchone()
                if data:
                    return Position(id=data[0], description=data[1])

class Query(graphene.ObjectType):
    allProfessionals = graphene.List(Professional)
    professional_by_id = graphene.Field(Professional, id=graphene.Int(required=True))
    allPositions = graphene.List(Position)
    position_by_id = graphene.Field(Position, id=graphene.Int(required=True))
    professionals_by_name = graphene.List(Professional, name=graphene.String(required=True))

    def resolve_allProfessionals(self, info):
        with psycopg2.connect(DATABASE_URL) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT id, name, position_id FROM djangoapp_professional")
                return [Professional(id=d[0], name=d[1], position_id=d[2]) for d in cur.fetchall()]

    def resolve_professional_by_id(self, info, id):
        with psycopg2.connect(DATABASE_URL) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT id, name, position_id FROM djangoapp_professional WHERE id = %s", (id,)
                )
                data = cur.fetchone()
                if data:
                    return Professional(id=data[0], name=data[1], position_id=data[2])

    def resolve_allPositions(self, info):
        with psycopg2.connect(DATABASE_URL) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT id, description FROM djangoapp_position")
                return [Position(id=d[0], description=d[1]) for d in cur.fetchall()]

    def resolve_professionals_by_name(self, info, name):
        with psycopg2.connect(DATABASE_URL) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT id, name, position_id FROM djangoapp_professional WHERE name LIKE %s",
                    (f'%{name}%',)
                )
                return [Professional(id=d[0], name=d[1], position_id=d[2]) for d in cur.fetchall()]

def get_schema():
    return graphene.Schema(query=Query)

What you’ve built

A Flask API serving a GraphQL endpoint backed by PostgreSQL. The schema defines Professional and Position types with multiple query entry points, each backed by a resolver that hits the database. Clients can now request exactly the fields they need in a single query.

Next steps

• Add mutations to create, update, and delete records. Graphene’s Mutation class follows the same resolver pattern as queries.
• Replace raw psycopg2 queries with SQLAlchemy and use graphene-sqlalchemy to auto-generate the GraphQL types from your ORM models.
• Add authentication middleware to protect sensitive queries, and use query depth limiting to prevent malicious clients from requesting deeply nested data.

Questions or feedback? Find me on LinkedIn or GitHub.