尝试使用Python的GraphQL库Strawberry

首先

在构建FastAPI的GraphQL服务器时，FastAPI的文档推荐使用名为Strawberry的库。为了学习GraphQL，我尝试了一下它。

草莓是什么意思？

这是一个面向Python的GraphQL库，我认为Graphene在Python中非常有名。它的设计思想类似于FastAPI，并且在使用FastAPI构建GraphQL服务器时推荐使用它（当然，你也可以使用其他库，如Graphene）。它不仅支持FastAPI，还支持其他各种框架，如Django、Flask和Chalice。

执行

我想以任务管理为主题，使用Strawberry来实现类似CRUD的功能。

文件结构

最终的文件结构如下所示。

.
├── main.py
└── tasks
    ├── __init__.py
    ├── inputs.py
    ├── models.py
    ├── repositories.py
    ├── resolvers.py
    ├── services.py
    └── types.py

安装

请安装以下两个库。

pip install 'strawberry-graphql[debug-server]'
pip install 'uvicorn[standard]'

准备工作

GraphQL是一种允许简单实现与许多不相关部分的技术。

import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional


class Status(Enum):
    TODO = "todo"
    DOING = "doing"
    DONE = "done"


@dataclass
class Task:
    title: str
    id: uuid.UUID = field(default_factory=uuid.uuid4)
    description: Optional[str] = None
    status: Status = Status.TODO
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)

from copy import deepcopy
from typing import Any

from tasks.models import Task


class InMemoryRepository:
    _store: dict[str, Any] = {}

    @classmethod
    def find_by_id(cls, id_: str) -> Task:
        task = cls._store.get(id_)
        if task is None:
            raise Exception("Not Found")

        return task

    @classmethod
    def find_all(cls) -> list[Task]:
        tasks = list(cls._store.values())

        return tasks

    @classmethod
    def save(cls, task: Task) -> None:
        cls._store[str(task.id)] = deepcopy(task)

    @classmethod
    def delete(cls, task: Task) -> None:
        del cls._store[str(task.id)]

from datetime import datetime
from typing import Optional

from tasks.models import Status, Task
from tasks.repositories import InMemoryRepository


class TaskService:
    def __init__(self, repo: type[InMemoryRepository]) -> None:
        self._repo = repo

    @property
    def repo(self) -> type[InMemoryRepository]:
        return self._repo

    def find(self, id: str) -> Task:
        task = self.repo.find_by_id(id)

        return task

    def find_all(self) -> list[Task]:
        tasks = self.repo.find_all()

        return tasks

    def create(self, *, title: str, description: Optional[str] = None) -> Task:
        task = Task(title=title, description=description)
        self.repo.save(task)

        return task

    def update(self, *, id: str, status: Status) -> Task:
        task = self.repo.find_by_id(id)
        task.status = status
        task.updated_at = datetime.utcnow()
        self.repo.save(task)

        return task

    def delete(self, id: str) -> Task:
        task = self.repo.find_by_id(id)
        self.repo.delete(task)

        return task

物体类型

在以上所述中，我们已完成前期准备工作。现在我们将开始实施与GraphQL相关的功能。
首先，我们要定义Object Type。有关GraphQL类型和Python类型的对应详细信息，请参阅此处。此外，我们将使用strawberry.enum来定义使用Enum定义的Status的类型。
通过将name=”Task”作为strawberry.type的参数，我们可以将其作为Task来处理。

from datetime import datetime
from typing import Optional

import strawberry

from tasks.models import Status, Task

StatusType = strawberry.enum(Status, name="Status")


@strawberry.type(name="Task")
class TaskType:
    id: strawberry.ID
    title: str
    description: Optional[str]
    status: StatusType
    created_at: datetime
    updated_at: datetime

    @classmethod
    def from_instance(cls, instance: Task) -> "TaskType":
        data = instance.__dict__

        return cls(**data)

输入类型

为了创建和更新任务，我们将使用strawberry.input来定义Input Type。

from typing import Optional

import strawberry

from tasks.types import StatusType


@strawberry.input
class AddTaskInput:
    title: str
    description: Optional[str] = None


@strawberry.input
class UpdateTaskInput:
    id: strawberry.ID
    status: StatusType

解决者 (jiě jué zhě)

定义一个Resolver。函数的参数将反映在GraphQL中。
对于add_task(task_input: AddTaskInput)情况，在GraphQL中将变为taskInput: AddTaskInput!，并且下划线形式将转换为驼峰式。

import strawberry

from tasks.inputs import AddTaskInput, UpdateTaskInput
from tasks.repositories import InMemoryRepository
from tasks.services import TaskService
from tasks.types import TaskType


def get_task(id: strawberry.ID) -> TaskType:
    db = InMemoryRepository
    service = TaskService(db)
    task = service.find(id)

    return TaskType.from_instance(task)


def get_tasks() -> list[TaskType]:
    db = InMemoryRepository
    service = TaskService(db)
    tasks = service.find_all()

    return [TaskType.from_instance(task) for task in tasks]


def add_task(task_input: AddTaskInput) -> TaskType:
    db = InMemoryRepository
    service = TaskService(db)
    task = service.create(**task_input.__dict__)

    return TaskType.from_instance(task)


def update_task(task_input: UpdateTaskInput) -> TaskType:
    db = InMemoryRepository
    service = TaskService(db)
    task = service.update(**task_input.__dict__)

    return TaskType.from_instance(task)


def delete_task(id: strawberry.ID) -> TaskType:
    db = InMemoryRepository
    service = TaskService(db)
    task = service.delete(id)

    return TaskType.from_instance(task)

查询与修改

最后，我们定义了Query和Mutation。这些定义的字段将会在GraphQL中反映出来。Snake Case将被转换为Camel Case，并且类型提示将成为返回值。

import strawberry

from tasks.resolvers import add_task, delete_task, get_task, get_tasks, update_task
from tasks.types import TaskType


@strawberry.type
class Query:
    task: TaskType = strawberry.field(resolver=get_task)
    tasks: list[TaskType] = strawberry.field(resolver=get_tasks)


@strawberry.type
class Mutation:
    task_add: TaskType = strawberry.field(resolver=add_task)
    task_update: TaskType = strawberry.field(resolver=update_task)
    task_delete: TaskType = strawberry.field(resolver=delete_task)


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

启动服务器

使用以下命令启动服务器。访问http://localhost:8000/graphql，将显示GraphiQL界面。

strawberry server main

结束

我們這次試著使用草莓構建GraphQL服務器。
由於草莓還處於開發初期，因此在使用時請先參考文件進行確認。

请参考

• Strawberry

• FastAPI

チュートリアル: GraphQL APIの設計