ایجاد Query ساده برای User و Task

در این مرحله، ابتدا Queryهای ساده‌ای تعریف می‌کنیم تا بتوانیم کاربران و تسک‌ها را از دیتابیس بخوانیم. هدف این است که یک پایه برای Mutationها فراهم کنیم. تغییرات فقط در بخش GraphQL اضافه می‌شوند و بقیه اپ نیاز به تغییر ندارد.

# graphql/users.py
import strawberry
from models.users import User as UserModel
from sqlalchemy.orm import Session
from typing import List

from graphql.tasks import TaskType  # برای نمایش تسک‌ها مرتبط با کاربر

@strawberry.type
class UserType:
    id: int
    email: str
    created_at: str
    updated_at: str | None
    tasks: list[TaskType] | None  # رابطه یک به چند با Task

@strawberry.type
class UserQuery:
    @strawberry.field
    def users(self, info) -> List[UserType]:
        db: Session = info.context["db"]
        users = db.query(UserModel).all()
        result = []
        for u in users:
            result.append(
                UserType(
                    id=u.id,
                    email=u.email,
                    created_at=str(u.created_at),
                    updated_at=str(u.updated_at),
                    tasks=[TaskType(
                        id=t.id,
                        title=t.title,
                        description=t.description,
                        is_completed=t.is_completed,
                        created_at=str(t.created_at),
                        updated_at=str(t.updated_at),
                        owner_id=t.owner_id
                    ) for t in u.tasks] if hasattr(u, "tasks") else []
                )
            )
        return result

# graphql/tasks.py
import strawberry
from models.tasks import Task as TaskModel
from sqlalchemy.orm import Session
from typing import List

@strawberry.type
class TaskType:
    id: int
    title: str
    description: str | None
    is_completed: bool
    created_at: str
    updated_at: str | None
    owner_id: int

@strawberry.type
class TaskQuery:
    @strawberry.field
    def tasks(self, info) -> List[TaskType]:
        db: Session = info.context["db"]
        tasks = db.query(TaskModel).all()
        return [
            TaskType(
                id=t.id,
                title=t.title,
                description=t.description,
                is_completed=t.is_completed,
                created_at=str(t.created_at),
                updated_at=str(t.updated_at),
                owner_id=t.owner_id
            ) for t in tasks
        ]

توضیح

• Query users تمام کاربران را برمی‌گرداند و شامل لیست تسک‌های هر کاربر است.
• Query tasks تمام تسک‌ها را برمی‌گرداند.
• در این مرحله هنوز Mutation نداریم، بنابراین داده‌ها فقط خوانده می‌شوند.

ایجاد Mutation ساده برای ثبت کاربر

# graphql/users.py (ادامه)
from models.users import User as UserModel
from config.database import SessionLocal

@strawberry.type
class UserMutation:
    @strawberry.mutation
    def create_user(self, info, email: str, password: str) -> UserType:
        db: Session = info.context["db"]
        new_user = UserModel(email=email, password=password)
        db.add(new_user)
        db.commit()
        db.refresh(new_user)
        return UserType(
            id=new_user.id,
            email=new_user.email,
            created_at=str(new_user.created_at),
            updated_at=str(new_user.updated_at),
            tasks=[]
        )

ایجاد Mutation ساده برای ثبت Task مرتبط با کاربر

# graphql/tasks.py (ادامه)
from models.tasks import Task as TaskModel
from config.database import SessionLocal

@strawberry.type
class TaskMutation:
    @strawberry.mutation
    def create_task(self, info, title: str, owner_id: int, description: str | None = None) -> TaskType:
        db: Session = info.context["db"]
        new_task = TaskModel(title=title, owner_id=owner_id, description=description)
        db.add(new_task)
        db.commit()
        db.refresh(new_task)
        return TaskType(
            id=new_task.id,
            title=new_task.title,
            description=new_task.description,
            is_completed=new_task.is_completed,
            created_at=str(new_task.created_at),
            updated_at=str(new_task.updated_at),
            owner_id=new_task.owner_id
        )

توضیح

• ابتدا یک کاربر ایجاد می‌کنیم تا Task بتواند به آن متصل شود.
• برای ایجاد Task، شناسه کاربر (owner_id) باید مشخص شود.
• در این مرحله Queryها برای خواندن داده‌ها و Mutationها برای ایجاد داده‌ها آماده هستند و می‌توان تست کرد.

نحوه اتصال GraphQL Schema به FastAPI

بعد از تعریف Query و Mutation برای User و Task، باید Schema نهایی را بسازیم و آن را به یک مسیر GraphQL در FastAPI متصل کنیم. در این بخش، فقط بخش‌های مرتبط با GraphQL تغییر می‌کنند و بقیه اپ (مثل دیتابیس یا مدل‌ها) نیازی به تغییر ندارد.

# schema.py
import strawberry
from graphql.users import UserQuery, UserMutation
from graphql.tasks import TaskQuery, TaskMutation

@strawberry.type
class Query(UserQuery, TaskQuery):
    pass

@strawberry.type
class Mutation(UserMutation, TaskMutation):
    pass

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

حالا این Schema را در FastAPI به GraphQLRouter متصل می‌کنیم:

# main.py
from strawberry.fastapi import GraphQLRouter
from schema import schema
from config.database import get_db

# اتصال GraphQL
graphql_app = GraphQLRouter(
    schema,
    context_getter=lambda request: {"db": next(get_db())}
)
app.include_router(graphql_app, prefix="/graphql")

@app.get("/")
def root():
    return {"message": "FastAPI + GraphQL is ready!"}

توضیح

• Router مربوط به GraphQL با GraphQLRouter ساخته می‌شود.
• Schema که شامل Query و Mutation است به Router داده می‌شود.
• با استفاده از context_getter، Session دیتابیس به Resolverها منتقل می‌شود.