Skip to content

Instantly share code, notes, and snippets.

@hasanisaeed

hasanisaeed/book.proto

Last active August 29, 2024 13:46
Show Gist options
  • Save hasanisaeed/7075a9b4db04a075a31ba84eb9a9a3b7 to your computer and use it in GitHub Desktop.
Save hasanisaeed/7075a9b4db04a075a31ba84eb9a9a3b7 to your computer and use it in GitHub Desktop.
Download ZIP
System design -> API Getways -> gRPC, GraphQL
Raw
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;
}
Raw
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))
Raw
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()
Raw
requirements.txt
fastapi
uvicorn
ariadne
# GraphQL
graphene
# gRPC
grpcio
grpcio-tools
@hasanisaeed
Copy link
Author

GraphQL:

  uvicorn graphql_app:app --reload

gRPC:

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment