Flask + GraphQL

APIs REST têm um problema de forma fixa: os clientes recebem dados demais ou precisam fazer várias requisições para obter o que precisam. O GraphQL inverte isso, o cliente declara exatamente o que quer e o servidor entrega exatamente isso. Sem campos extras, sem N+1 requisições para juntar dados relacionados.

TL;DR: Configure uma API Flask + GraphQL com graphene e psycopg2, definindo tipos, consultas e resolvers conectados a um banco PostgreSQL.
Stack: Python, Flask, GraphQL, graphene, psycopg2, PostgreSQL
Nível: Intermediário
Tempo de leitura: ~8 min

Instalar libs

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

Criar 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)

Implementar schemas.py

Esse arquivo define todos os seus tipos GraphQL, a classe Query e os métodos resolvers que buscam dados do banco. Cada campo de um tipo que precisa de uma consulta ao banco ganha seu próprio método resolve_*. Por convenção, cada atributo tem uma implementação correspondente com o prefixo “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)

O que você construiu

Uma API Flask servindo um endpoint GraphQL com PostgreSQL. O schema define os tipos Professional e Position com vários pontos de entrada de consulta, cada um com um resolver que acessa o banco. Os clientes agora podem solicitar exatamente os campos que precisam em uma única consulta.

Próximos passos

  • Adicione mutations para criar, atualizar e deletar registros. A classe Mutation do graphene segue o mesmo padrão de resolver das queries.
  • Substitua as queries brutas do psycopg2 pelo SQLAlchemy e use o graphene-sqlalchemy para gerar automaticamente os tipos GraphQL a partir dos seus modelos ORM.
  • Adicione middleware de autenticação para proteger consultas sensíveis, e use limitação de profundidade de consulta para evitar que clientes maliciosos peçam dados profundamente aninhados.

Dúvidas ou feedback? Me encontre no LinkedIn ou GitHub.

Deixe um comentário