Select Git revision
items.py 3.43 KiB
import logging
import grpc
from perxis.items import items_pb2, items_pb2_grpc
from google.protobuf.struct_pb2 import Struct
ACCESS_TOKEN = 'Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik5UWXhSVE5FTnprek5qSkVPRFkyTmpORk9VVXhRVEUzUmpWRVFqY3pRVE5CTjBJeE1ETTNOUSJ9.eyJodHRwczovL2hvb3AucGVyeC5ydS9pZGVudGl0eSI6eyJncm91cHMiOltdfSwibmlja25hbWUiOiJnb3NoaWstZSIsIm5hbWUiOiJHZW9yZ2l5IEV0ZXJldnNraXkiLCJwaWN0dXJlIjoiaHR0cHM6Ly9hdmF0YXJzLmdpdGh1YnVzZXJjb250ZW50LmNvbS91Lzc2NjAzMzE_dj00IiwidXBkYXRlZF9hdCI6IjIwMjEtMDQtMjBUMDc6MDM6NTAuMzM5WiIsImlzcyI6Imh0dHBzOi8vYWR3ei5hdXRoMC5jb20vIiwic3ViIjoiZ2l0aHVifDc2NjAzMzEiLCJhdWQiOiJ6d3JaaGZrd0owNlQ3MlE4ZXhOZHRFS0NxTzBja0ZpWCIsImlhdCI6MTYxODkwMjIzMCwiZXhwIjoxNjE4OTM4MjMwLCJhdXRoX3RpbWUiOjE2MTg5MDIyMzAsImF0X2hhc2giOiJNZU1FdkxBRm14N1JXZ3pDaEwxdlBBIiwibm9uY2UiOiJrTEExZDZhM0V4cjBDMlU5UVBoNG5DUkhiQ3ZvcU1TdiJ9.BqkhCwRi7_LZ0FANXU9fe_9MP0lIq7TewBo3OiSe92FXSXmyKMIEeUPnPtp0oUzf6taFgVCyQJiKrbJfpnlZYK4rVetPEWKvGpMiyv4IsQ_6Y7ZTBKdOFZrej0sKB9EFUjkBX01g4ihDyLUaz3xJYZX4qLApEPS4ubAFTuqsa_M1-rI9_8cXa3leMnQHqEOkL2Cb-N2M2uWrXYYzch8PpVyTMiBK39CA8-EJ1z8MGwjcKsqR8wryG4OWCAn84i8kt7wFAad9Tjb6YsCp1v8HlZR4cUj7u2-ztEzHDs5BNbG2dyloizBqh1XScCA261asybQgYDJDEBmqztk9mKgsVg'
SPACE_ID = 'c1mrvup3e6litup7tf0g'
ENV_ID = 'master'
COLL_ID = 'test_eterevskiy'
def create(stub, space_id, env_id, coll_id, data):
return stub.Create.with_call(
items_pb2.CreateRequest(item=items_pb2.Item(
space_id=space_id, env_id=env_id, collection_id=coll_id, data=data
)),
metadata=(
('authorization', f'{ACCESS_TOKEN}'),
)
)
def get(stub, space_id, env_id, coll_id, item_id):
return stub.Get.with_call(
items_pb2.GetRequest(
space_id=space_id, env_id=env_id, collection_id=coll_id, item_id=item_id
),
metadata=(
('authorization', f'{ACCESS_TOKEN}'),
)
)
def find(stub, space_id, env_id, coll_id, filter=None):
return stub.Find.with_call(
items_pb2.FindRequest(space_id=space_id, env_id=env_id, collection_id=coll_id, filter=filter),
metadata=(
('authorization', f'{ACCESS_TOKEN}'),
)
)
def update(stub, space_id, env_id, coll_id, item_id, data):
return stub.Update.with_call(
items_pb2.UpdateRequest(
item=items_pb2.Item(space_id=space_id, env_id=env_id, collection_id=coll_id, id=item_id, data=data)
),
metadata=(
('authorization', f'{ACCESS_TOKEN}'),
)
)
def main() -> None:
with grpc.insecure_channel('envoy.perxis.pt.perx.ru:80') as channel:
stub = items_pb2_grpc.ItemsStub(channel)
print('---------- CREATE ----------')
data = Struct()
data.update({'title': 'test'})
response, state = create(stub, SPACE_ID, ENV_ID, COLL_ID, data)
created = response.created
print(response.created, state)
print('---------- GET ----------')
response, state = get(stub, SPACE_ID, ENV_ID, COLL_ID, created.id)
print(response.item, state)
print('---------- FIND ----------')
response, state = find(stub, SPACE_ID, ENV_ID, COLL_ID)
print(response.items, state)
print('---------- UPDATE ----------')
data = Struct()
data.update({'title': 'New test title'})
response, state = update(stub, SPACE_ID, ENV_ID, COLL_ID, created.id, data)
print(state)
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
main()