Este artículo explora el uso de Protocol Buffers y gRPC en Python, cubriendo desde la definciión de mensajes hasta la creación de servicios RPC.
- Uso de Protocol Buffers en Python
1.1 Instalación de Protocol Buffers
Para comenzar, necesitas instalar las bibliotecas necesarias para gRPC y sus herramientas.
pip install grpcio
pip install grpcio-tools
1.2 Definición de formato en proto3
Se define la estructura de los mensajes utilizando la sintaxis de Protocol Buffers version 3. Crearemos un archivo llamado hello.proto.
syntax = "proto3";
message HelloRequest {
string name = 1;
}
1.3 Generación de archivos Python a partir de proto
Utiliza la herramienta grpc_tools.protoc para generar el código Python que permitirá la serialización y deserialización de tus mensajes.
Navega al directorio donde se encuentra tu archivo hello.proto y ejecuta el siguiente comando:
python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. hello.proto
Este comando generará dos archivos Python:
--python_out=.: Archivo para la serialización/deserialización básica de mensajes.--grpc_python_out=.: Archivo que contiene la estructura para el cliente y servidor gRPC.
1.4 Comparativa del resultado de la generación de proto
Protocol Buffers ofrece una serialización binaria eficiente, resultando en datos más compactos en comparación con formatos como JSON.
import json
from google.protobuf.message import Message # Asumiendo que el archivo proto se llama hello_pb2.py
# Reemplaza 'your_proto_package.hello_pb2' con la ruta correcta a tu archivo generado
# Ejemplo: if your proto is in a 'proto' directory, it might be 'proto.hello_pb2'
# For demonstration, let's assume a direct import:
try:
import hello_pb2
except ImportError:
print("Por favor, genera los archivos proto primero con el comando adecuado.")
# Placeholder for demonstration if files are not generated
class MockHelloRequest:
def __init__(self):
self.name = ""
def SerializeToString(self):
return f"\\n\\x04{self.name}".encode('utf-8') # Mock representation
def ParseFromString(self, data):
parts = data.decode('utf-8').split("4")
if len(parts) > 1:
self.name = parts[1]
hello_pb2 = type('obj', (object,), {'HelloRequest': MockHelloRequest})()
def main():
request_obj = hello_pb2.HelloRequest()
request_obj.name = "jeff"
serialized_data = request_obj.SerializeToString()
print(f"Datos serializados (protobuf): {serialized_data}")
print(f"Longitud serializada (protobuf): {len(serialized_data)}")
deserialized_obj = hello_pb2.HelloRequest()
deserialized_obj.ParseFromString(serialized_data)
print(f"Nombre deserializado: {deserialized_obj.name}")
json_data = {"name": "jeff"}
json_string = json.dumps(json_data)
print(f"Datos serializados (JSON): {json_string}")
print(f"Longitud serializada (JSON): {len(json_string)}")
if __name__ == '__main__':
main()
- Uso de gRPC en Python
2.1 Creación del archivo hello.proto
Definimos un servicio gRPC llamado Greeter con un método SayHello. Este servicio utiliza los mensajes HelloRequest y HelloReply.
syntax = "proto3";
package services;
# Opcional: especifica el paquete Go si planeas usarlo en Go también
option go_package = "./;proto";
service Greeter {
rpc SayHello (HelloRequest) returns (HelloReply);
}
message HelloRequest {
string name = 1;
}
message HelloReply {
string message = 1;
}
2.2 Generación de archivos Python a partir de proto
El proceso es similar al anterior, pero esta vez generamos los archivos para el servicio gRPC.
Ejecuta el comando en el directorio del archivo hello.proto:
python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. hello.proto
Nota importante: Es posible que necesites ajustar las declaraciones de importación en los archivos generados (especialmente en el archivo *_grpc.py) para que Python pueda encontrar correctamente los módulos generados (por ejemplo, hello_pb2).
2.3 Implementación del lado del servidor
El servidor gRPC implementa la lógica del servicio definido en el archivo .proto.
import grpc
from concurrent import futures
# Asegúrate de que la ruta de importación sea correcta según tu estructura de proyecto
# Ejemplo: from your_project_root.proto import hello_pb2, hello_pb2_grpc
try:
import hello_pb2
import hello_pb2_grpc
except ImportError:
print("Asegúrate de que los archivos proto han sido generados y las rutas de importación son correctas.")
# Placeholder classes for demonstration if generation failed
class MockHelloRequest: pass
class MockHelloReply:
def __init__(self, message=""): self.message = message
class MockGreeterServicer:
def SayHello(self, request, context): return MockHelloReply(message="Mock reply")
class MockGreeterStub:
def __init__(self, channel): pass
def SayHello(self, request): return MockHelloReply(message="Mock reply")
hello_pb2 = type('obj', (object,), {'HelloRequest': MockHelloRequest, 'HelloReply': MockHelloReply})()
hello_pb2_grpc = type('obj', (object,), {'GreeterServicer': MockGreeterServicer, 'add_GreeterServicer_to_server': lambda servicer, server: None, 'GreeterStub': MockGreeterStub})()
class GreeterService(hello_pb2_grpc.GreeterServicer):
def SayHello(self, request, context):
"""Implementa la lógica del método SayHello."""
return hello_pb2.HelloReply(message=f"Hola, {request.name}")
def start_server():
"""Inicia el servidor gRPC."""
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
hello_pb2_grpc.add_GreeterServicer_to_server(GreeterService(), server)
server_address = '127.0.0.1:50051' # Puerto común para gRPC
server.add_insecure_port(server_address)
server.start()
print(f"Servidor gRPC iniciado en {server_address}")
server.wait_for_termination()
if __name__ == '__main__':
start_server()
2.4 Implementación del lado del cliente
El cliente gRPC se conecta al servidor y llama a los métodos del servicio.
import grpc
# Asegúrate de que la ruta de importación sea correcta según tu estructura de proyecto
# Ejemplo: from your_project_root.proto import hello_pb2, hello_pb2_grpc
try:
import hello_pb2
import hello_pb2_grpc
except ImportError:
print("Asegúrate de que los archivos proto han sido generados y las rutas de importación son correctas.")
# Placeholder classes for demonstration if generation failed
class MockHelloRequest:
def __init__(self, name=""): self.name = name
class MockHelloReply:
def __init__(self, message=""): self.message = message
class MockGreeterStub:
def __init__(self, channel): pass
def SayHello(self, request): return MockHelloReply(message="Mock reply from client")
hello_pb2 = type('obj', (object,), {'HelloRequest': MockHelloRequest})()
hello_pb2_grpc = type('obj', (object,), {'GreeterStub': MockGreeterStub})()
def make_rpc_call():
"""Realiza una llamada RPC al servicio Greeter."""
server_address = '127.0.0.1:50051' # Debe coincidir con la dirección del servidor
# Utiliza 'with' para asegurar que el canal se cierre automáticamente
with grpc.insecure_channel(server_address) as channel:
stub = hello_pb2_grpc.GreeterStub(channel)
try:
response = stub.SayHello(hello_pb2.HelloRequest(name="Mundo"))
return response.message
except grpc.RpcError as e:
print(f"Error en la llamada RPC: {e.code()} - {e.details()}")
return None
if __name__ == '__main__':
reply_message = make_rpc_call()
if reply_message:
print(f"Respuesta del servidor: {reply_message}")