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.