Creare un'API GraphQL in Python

  • Python
  • Gabriele Romanato

Questo articolo mostra un percorso chiaro per esporre un singolo modello, Post, tramite GraphQL in Python. Useremo FastAPI per il server HTTP e Strawberry per GraphQL. Copriremo schema, resolver, paginazione “page-based”, ordinamento per publishedAt, validazione con Pydantic, autorizzazione tramite context e CORS sull'endpoint /graphql.

Stack e struttura

  • Server HTTP: FastAPI
  • GraphQL: Strawberry
  • Validazione: Pydantic
  • Persistenza: repository (in-memory per l'esempio; facilmente sostituibile con SQL/NoSQL)
  • CORS: middleware nativo di FastAPI

Schema GraphQL: tipo Post e operazioni

Uno schema essenziale con tipo Post, wrapper di paginazione e operazioni CRUD.

"""Rappresenta un contenuto pubblicato."""
type Post {
  id: ID!
  title: String!
  body: String!
  publishedAt: String
  createdAt: String!
  updatedAt: String!
}

"""Metadati di paginazione (page-based)."""
type PaginatorInfo {
  count: Int!
  currentPage: Int!
  lastPage: Int!
  perPage: Int!
  total: Int!
  hasMorePages: Boolean!
}

"""Risultato paginato dei Post."""
type PostPaginator {
  data: [Post!]!
  paginatorInfo: PaginatorInfo!
}

type Query {
  posts(page: Int = 1, perPage: Int = 10, sort: String = "desc"): PostPaginator!
  post(id: ID!): Post
}

type Mutation {
  createPost(title: String!, body: String!, publishedAt: String): Post!
  updatePost(id: ID!, title: String, body: String, publishedAt: String): Post!
  deletePost(id: ID!): Boolean!
}

Dominio e repository

Definiamo l'entità di dominio e l'interfaccia del repository. In seguito potrai sostituire l'implementazione con SQLAlchemy o un driver NoSQL.

# domain.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Protocol, Tuple, List, Dict, Any


@dataclass
class PostEntity:
    id: str
    title: str
    body: str
    published_at: Optional[datetime]
    created_at: datetime
    updated_at: datetime


class PostRepository(Protocol):
    def list(self, page: int, per_page: int, sort_asc: bool) -> Tuple[List[PostEntity], int]: ...
    def find_by_id(self, post_id: str) -> Optional[PostEntity]: ...
    def create(self, data: Dict[str, Any]) -> PostEntity: ...
    def update(self, post_id: str, patch: Dict[str, Any]) -> PostEntity: ...
    def delete(self, post_id: str) -> bool: ...

Tipi GraphQL con Strawberry

Mappiamo i campi Python in camelCase per l'API GraphQL.

# schema.py
from __future__ import annotations

from typing import List, Optional
import strawberry


@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    body: str
    created_at: str = strawberry.field(name="createdAt")
    updated_at: str = strawberry.field(name="updatedAt")
    published_at: Optional[str] = strawberry.field(name="publishedAt", default=None)


@strawberry.type
class PaginatorInfo:
    count: int
    current_page: int = strawberry.field(name="currentPage")
    last_page: int = strawberry.field(name="lastPage")
    per_page: int = strawberry.field(name="perPage")
    total: int
    has_more_pages: bool = strawberry.field(name="hasMorePages")


@strawberry.type
class PostPaginator:
    data: List[Post]
    paginator_info: PaginatorInfo = strawberry.field(name="paginatorInfo")

Mapper: dal dominio al tipo GraphQL

# mappers.py
from __future__ import annotations

from datetime import datetime
from typing import Optional
from domain import PostEntity
from schema import Post


def _to_iso(dt: Optional[datetime]) -> Optional[str]:
    if not dt:
        return None
    s = dt.astimezone().isoformat(timespec="seconds")
    return s.replace("+00:00", "Z")


def map_post(entity: PostEntity) -> Post:
    return Post(
        id=entity.id,
        title=entity.title,
        body=entity.body,
        created_at=_to_iso(entity.created_at) or "",
        updated_at=_to_iso(entity.updated_at) or "",
        published_at=_to_iso(entity.published_at),
    )

Resolver: Query e Mutation

Resolver semplici: validano input, applicano l'autorizzazione tramite context, delegano al repository e costruiscono i metadati di paginazione.

# resolvers.py
from __future__ import annotations

from datetime import datetime, timezone
from typing import Optional

import strawberry
from pydantic import BaseModel, Field, field_validator
from strawberry.types import Info

from mappers import map_post
from schema import Post, PostPaginator, PaginatorInfo


class CreatePostModel(BaseModel):
    title: str = Field(min_length=1, max_length=255)
    body: str = Field(min_length=1)
    publishedAt: Optional[str] = None

    @field_validator("publishedAt")
    @classmethod
    def _validate_dt(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        datetime.fromisoformat(v.replace("Z", "+00:00"))
        return v


class UpdatePostModel(BaseModel):
    id: str
    title: Optional[str] = Field(default=None, min_length=1, max_length=255)
    body: Optional[str] = Field(default=None, min_length=1)
    publishedAt: Optional[str] = None

    @field_validator("publishedAt")
    @classmethod
    def _validate_dt(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        datetime.fromisoformat(v.replace("Z", "+00:00"))
        return v


def _require_user(info: Info) -> None:
    if not info.context.get("user"):
        raise PermissionError("Unauthorized")


@strawberry.type
class Query:
    @strawberry.field
    def posts(self, info: Info, page: int = 1, per_page: int = 10, sort: str = "desc") -> PostPaginator:
        repo = info.context["repo"]
        asc = str(sort).lower() == "asc"

        items, total = repo.list(page=page, per_page=per_page, sort_asc=asc)
        last_page = max(1, (total + per_page - 1) // per_page)

        return PostPaginator(
            data=[map_post(p) for p in items],
            paginator_info=PaginatorInfo(
                count=len(items),
                current_page=page,
                last_page=last_page,
                per_page=per_page,
                total=total,
                has_more_pages=page < last_page,
            ),
        )

    @strawberry.field
    def post(self, info: Info, id: strawberry.ID) -> Optional[Post]:
        repo = info.context["repo"]
        found = repo.find_by_id(str(id))
        return map_post(found) if found else None


@strawberry.type
class Mutation:
    @strawberry.field
    def create_post(
        self,
        info: Info,
        title: str,
        body: str,
        published_at: Optional[str] = strawberry.UNSET,
    ) -> Post:
        _require_user(info)

        payload = CreatePostModel(
            title=title,
            body=body,
            publishedAt=published_at if published_at is not strawberry.UNSET else None,
        )

        now = datetime.now(timezone.utc)
        pub = (
            datetime.fromisoformat(payload.publishedAt.replace("Z", "+00:00"))
            if payload.publishedAt
            else None
        )

        created = info.context["repo"].create(
            {
                "title": payload.title,
                "body": payload.body,
                "published_at": pub,
                "created_at": now,
                "updated_at": now,
            }
        )
        return map_post(created)

    @strawberry.field
    def update_post(
        self,
        info: Info,
        id: strawberry.ID,
        title: Optional[str] = None,
        body: Optional[str] = None,
        published_at: Optional[str] = None,
    ) -> Post:
        _require_user(info)

        payload = UpdatePostModel(
            id=str(id), title=title, body=body, publishedAt=published_at
        )

        patch = {}
        if payload.title is not None:
            patch["title"] = payload.title
        if payload.body is not None:
            patch["body"] = payload.body
        if payload.publishedAt is not None:
            patch["published_at"] = (
                datetime.fromisoformat(payload.publishedAt.replace("Z", "+00:00"))
                if payload.publishedAt
                else None
            )
        patch["updated_at"] = datetime.now(timezone.utc)

        updated = info.context["repo"].update(payload.id, patch)
        return map_post(updated)

    @strawberry.field
    def delete_post(self, info: Info, id: strawberry.ID) -> bool:
        _require_user(info)
        return info.context["repo"].delete(str(id))

Server FastAPI e CORS su /graphql

Montiamo il router GraphQL, abilitiamo CORS e passiamo nel context sia l'utente autenticato sia il repository.

# app.py
from __future__ import annotations

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from strawberry.fastapi import GraphQLRouter
import strawberry

from resolvers import Query, Mutation


schema = strawberry.Schema(query=Query, mutation=Mutation)


async def get_context(req: Request) -> dict:
    token = req.headers.get("authorization", "")
    user = {"id": "u1"} if token else None
    repo = req.app.state.repo
    return {"user": user, "repo": repo}


graphql_app = GraphQLRouter(schema, context_getter=get_context)

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "https://app.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["authorization", "content-type"],
    max_age=600,
)

app.include_router(graphql_app, prefix="/graphql")


@app.on_event("startup")
async def startup() -> None:
    from repo_inmemory import InMemoryPostRepo

    app.state.repo = InMemoryPostRepo()

Repository in-memory (sostituibile con DB)

Ordinamento per published_at (con fallback) e paginazione per pagina.

# repo_inmemory.py
from __future__ import annotations

from datetime import datetime
from typing import List, Tuple, Dict, Any, Optional
import uuid

from domain import PostEntity, PostRepository


class InMemoryPostRepo(PostRepository):  # type: ignore[misc]
    def __init__(self) -> None:
        self._rows: Dict[str, PostEntity] = {}

    def list(self, page: int, per_page: int, sort_asc: bool) -> Tuple[List[PostEntity], int]:
        rows = list(self._rows.values())
        rows.sort(
            key=lambda r: (r.published_at or datetime.min, r.id),
            reverse=not sort_asc,
        )
        total = len(rows)
        start = max(0, (page - 1) * per_page)
        end = start + per_page
        return rows[start:end], total

    def find_by_id(self, post_id: str) -> Optional[PostEntity]:
        return self._rows.get(post_id)

    def create(self, data: Dict[str, Any]) -> PostEntity:
        pid = str(uuid.uuid4())
        entity = PostEntity(
            id=pid,
            title=data["title"],
            body=data["body"],
            published_at=data.get("published_at"),
            created_at=data["created_at"],
            updated_at=data["updated_at"],
        )
        self._rows[pid] = entity
        return entity

    def update(self, post_id: str, patch: Dict[str, Any]) -> PostEntity:
        cur = self._rows.get(post_id)
        if not cur:
            raise KeyError("not found")
        for k, v in patch.items():
            setattr(cur, k, v)
        self._rows[post_id] = cur
        return cur

    def delete(self, post_id: str) -> bool:
        return self._rows.pop(post_id, None) is not None

Esempi dal punto di vista del client

Le query richiedono solo i campi necessari; le mutation ritornano l'oggetto aggiornato oppure un boolean per l'eliminazione (niente subselection su Boolean!).

query {
  posts(page: 1, perPage: 5, sort: "desc") {
    data { id title publishedAt }
    paginatorInfo { currentPage lastPage perPage total hasMorePages count }
  }
}

query {
  post(id: "123") { id title body publishedAt }
}

mutation {
  createPost(title: "Titolo", body: "Testo", publishedAt: "2025-01-01T00:00:00Z") {
    id title publishedAt
  }
}

mutation {
  updatePost(id: "123", title: "Nuovo titolo") { id title updatedAt }
}

mutation {
  deletePost(id: "123")
}

Note su validazione, autorizzazione, CORS

  • Validazione: Pydantic centralizza le regole (lunghezze, formati, date ISO 8601) e semplifica i messaggi di errore.
  • Autorizzazione: risolvi l'identità a livello di middleware e passa l'utente nel context dei resolver.
  • CORS: abilita il preflight OPTIONS, includi authorization negli header ammessi, elenca gli origin reali quando usi credenziali.

Conclusione

Con pochi componenti mirati si ottiene un'API GraphQL in Python focalizzata su Post: schema pulito, resolver snelli, paginazione e ordinamento prevedibili, validazione e autorizzazione esplicite e CORS configurato per scenari reali. Questa base è facile da estendere ad altri modelli mantenendo coerenza e qualità del codice.

Torna su