System design -> API Getways -> gRPC, GraphQL

book.proto

syntax = "proto3";

service BookService {
  rpc GetBooks (Empty) returns (BookList);
  rpc GetBookByTitle (BookTitle) returns (Book);
  rpc AddBook (Book) returns (Book);
  rpc UpdateBook (UpdateBookRequest) returns (Book);
  rpc DeleteBook (BookTitle) returns (DeleteBookResponse);
}

message Empty {}

message Book {
  string title = 1;
  string author = 2;
}

message BookTitle {
  string title = 1;
}

message UpdateBookRequest {
  string title = 1;
  string newTitle = 2;
  string newAuthor = 3;
}

message BookList {
  repeated Book books = 1;
}

message DeleteBookResponse {
  bool success = 1;
}

grapql_app.py

from fastapi import FastAPI
from ariadne import QueryType, MutationType, make_executable_schema, graphql_sync
from ariadne.asgi import GraphQL
from fastapi.middleware.cors import CORSMiddleware

# تعریف نوع‌های GraphQL
type_defs = """
    type Query {
        books: [Book]
        book(title: String!): Book
    }

    type Book {
        title: String!
        author: String!
    }

    type Mutation {
        addBook(title: String!, author: String!): Book
        updateBook(title: String!, newTitle: String, newAuthor: String): Book
        deleteBook(title: String!): Boolean
    }
"""

# ساخت کوئری‌ها
query = QueryType()
mutation = MutationType()

# لیست کتاب‌ها
books = [
    {"title": "Fluent Python", "author": "Luciano Ramalho"},
    {"title": "Python Crash Course", "author": "Eric Matthes"},
    {"title": "Effective Python", "author": "Brett Slatkin"},
    {"title": "Automate the Boring Stuff with Python", "author": "Al Sweigart"},
    {"title": "Learning Python", "author": "Mark Lutz"}
]


# تعریف کوئری‌های پیچیده‌تر

# بازیابی تمام کتاب‌ها
@query.field("books")
def resolve_books(*_):
    return books


# پیدا کردن کتاب بر اساس عنوان
@query.field("book")
def resolve_book(_, info, title):
    for book in books:
        if book["title"] == title:
            return book
    return None


# تعریف Mutation‌ها

# اضافه کردن کتاب جدید
@mutation.field("addBook")
def resolve_add_book(_, info, title, author):
    new_book = {"title": title, "author": author}
    books.append(new_book)
    return new_book


# بروزرسانی اطلاعات کتاب
@mutation.field("updateBook")
def resolve_update_book(_, info, title, newTitle=None, newAuthor=None):
    for book in books:
        if book["title"] == title:
            if newTitle:
                book["title"] = newTitle
            if newAuthor:
                book["author"] = newAuthor
            return book
    return None


# حذف کتاب
@mutation.field("deleteBook")
def resolve_delete_book(_, info, title):
    global books
    for book in books:
        if book["title"] == title:
            books = [b for b in books if b["title"] != title]
            return True
    return False


# ساخت schema اجرایی
schema = make_executable_schema(type_defs, query, mutation)

# ساخت اپلیکیشن FastAPI
app = FastAPI()

# فعال کردن CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# افزودن endpoint GraphQL
app.add_route("/graphql", GraphQL(schema, debug=True))

grpc_app.py

from concurrent import futures
import grpc
import book_pb2
import book_pb2_grpc

# لیست کتاب‌ها
books = [
    {"title": "Fluent Python", "author": "Luciano Ramalho"},
    {"title": "Python Crash Course", "author": "Eric Matthes"},
    {"title": "Effective Python", "author": "Brett Slatkin"},
    {"title": "Automate the Boring Stuff with Python", "author": "Al Sweigart"},
    {"title": "Learning Python", "author": "Mark Lutz"}
]


class BookServiceServicer(book_pb2_grpc.BookServiceServicer):

    def GetBooks(self, request, context):
        book_list = book_pb2.BookList()
        for book in books:
            book_list.books.add(title=book['title'], author=book['author'])
        return book_list

    def GetBookByTitle(self, request, context):
        for book in books:
            if book['title'] == request.title:
                return book_pb2.Book(title=book['title'], author=book['author'])
        context.set_code(grpc.StatusCode.NOT_FOUND)
        context.set_details('Book not found')
        return book_pb2.Book()

    def AddBook(self, request, context):
        new_book = {"title": request.title, "author": request.author}
        books.append(new_book)
        return book_pb2.Book(title=new_book['title'], author=new_book['author'])

    def UpdateBook(self, request, context):
        for book in books:
            if book['title'] == request.title:
                if request.newTitle:
                    book['title'] = request.newTitle
                if request.newAuthor:
                    book['author'] = request.newAuthor
                return book_pb2.Book(title=book['title'], author=book['author'])
        context.set_code(grpc.StatusCode.NOT_FOUND)
        context.set_details('Book not found')
        return book_pb2.Book()

    def DeleteBook(self, request, context):
        global books
        for book in books:
            if book['title'] == request.title:
                books = [b for b in books if b['title'] != request.title]
                return book_pb2.DeleteBookResponse(success=True)
        return book_pb2.DeleteBookResponse(success=False)


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    book_pb2_grpc.add_BookServiceServicer_to_server(BookServiceServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("gRPC server started on port 50051")
    server.wait_for_termination()


if __name__ == '__main__':
    serve()

requirements.txt

fastapi
uvicorn
ariadne

# GraphQL
graphene

# gRPC
grpcio
grpcio-tools

GraphQL:

  uvicorn graphql_app:app --reload

gRPC:

  python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. book.proto
  
  python grpc_app.py