Skip to content

Instantly share code, notes, and snippets.

@wfng92
Created September 8, 2022 08:59
Show Gist options
  • Save wfng92/0597609c4e16de74c4a39f4c73e067c7 to your computer and use it in GitHub Desktop.
Save wfng92/0597609c4e16de74c4a39f4c73e067c7 to your computer and use it in GitHub Desktop.
import logging
import os
import grpc
from protos import hello_pb2, hello_pb2_grpc
def get_filepath(filename, extension):
return f'{filename}{extension}'
def read_iterfile(filepath, chunk_size=1024):
split_data = os.path.splitext(filepath)
filename = split_data[0]
extension = split_data[1]
metadata = hello_pb2.MetaData(filename=filename, extension=extension)
yield hello_pb2.UploadFileRequest(metadata=metadata)
with open(filepath, mode="rb") as f:
while True:
chunk = f.read(chunk_size)
if chunk:
entry_request = hello_pb2.UploadFileRequest(chunk_data=chunk)
yield entry_request
else: # The chunk was empty, which means we're at the end of the file
return
def run():
with grpc.insecure_channel('localhost:50051') as channel:
stub = hello_pb2_grpc.GreeterStub(channel)
response = stub.SayHello(hello_pb2.HelloRequest(name='John Doe', age=30))
print("Greeter client received: " + response.message)
response = stub.UploadFile(read_iterfile('test.txt'))
print("Greeter client received: " + response.message)
filename = 'test'
extension = '.jpg'
filepath = get_filepath(filename, extension)
for entry_response in stub.DownloadFile(hello_pb2.MetaData(filename=filename, extension=extension)):
with open(filepath, mode="ab") as f:
f.write(entry_response.chunk_data)
if __name__ == '__main__':
logging.basicConfig()
run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment