Skip to content
Snippets Groups Projects
Select Git revision
  • 36f2d432bfda9ea5ece31c9ff81cc54ca2a1f625
  • master default protected
  • bugfix/fix-return-var-in-find
  • feature/upgrade2
  • v1.10.0
  • v1.8.2
  • v1.8.1
  • v1.8.0
  • 1.7.3
  • v1.7.1
  • v1.6.1
  • v1.6.0
  • v1.5.0
  • v1.4.1
  • v1.3.0
  • v1.2.2
  • v1.2.1
  • v1.2.0
  • v1.0.1
  • v1.0.0
  • v0.0.23
  • v0.0.17
  • v0.0.10
  • v0.0.9
24 results

spaces.py

Blame
  • 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()