environments.py

import logging

import grpc

from perxis.environments import environments_pb2, environments_pb2_grpc

ACCESS_TOKEN = 'Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Ik5UWXhSVE5FTnprek5qSkVPRFkyTmpORk9VVXhRVEUzUmpWRVFqY3pRVE5CTjBJeE1ETTNOUSJ9.eyJodHRwczovL2hvb3AucGVyeC5ydS9pZGVudGl0eSI6eyJncm91cHMiOltdfSwibmlja25hbWUiOiJnb3NoaWstZSIsIm5hbWUiOiJHZW9yZ2l5IEV0ZXJldnNraXkiLCJwaWN0dXJlIjoiaHR0cHM6Ly9hdmF0YXJzLmdpdGh1YnVzZXJjb250ZW50LmNvbS91Lzc2NjAzMzE_dj00IiwidXBkYXRlZF9hdCI6IjIwMjEtMDYtMTFUMTI6MDU6MDkuMjQ3WiIsImlzcyI6Imh0dHBzOi8vYWR3ei5hdXRoMC5jb20vIiwic3ViIjoiZ2l0aHVifDc2NjAzMzEiLCJhdWQiOiJ6d3JaaGZrd0owNlQ3MlE4ZXhOZHRFS0NxTzBja0ZpWCIsImlhdCI6MTYyMzQxMzEwOSwiZXhwIjoxNjIzNDQ5MTA5LCJhdXRoX3RpbWUiOjE2MjM0MTMxMDksImF0X2hhc2giOiIwVTNrRmJTTXpnWllWSVlKTHpMUWNBIiwibm9uY2UiOiJxTGxKZmVMSUtBOUtoMVRoWXE3UkxLSkxCMjF-X29TZCJ9.Z3BYpwyAUQYt5gyGyQSD14Lj5zwegYqfyaA-tsgVrHYdyYzb8VCG8Tb_CbZ6zjpRGpcOSuDCgPo6b8METTUoHTBgWyPgaYXLWeoecE4FjRrFgonL0Yltg0BAMG8hACxTkLsg3q3s1pA0FdUzNesDPRWgoyTQehh7ODwFCV9FLpV3aV1RYwqj3I_qMsfATCoDsn2_msozB2asge9PsH8cblQj5ZdT2JEFTU0ZLI-MFNu6aubpHd3WIgMXmfgmDASrsARx-W59enwZSls01Y3WRLfoQVZ8YqjdnOIXeolfkpr32DSuzcgOP_oxylXSa_grMlCbRBOZd1Z84teYteNU0g'
SPACE_ID = 'c2qcp9cuaccmpj8lmom0'


def create(stub, space_id, description):
    return stub.Create.with_call(
        environments_pb2.CreateRequest(env=environments_pb2.Environment(
            space_id=space_id,
            description=description
        )),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def get(stub, space_id, env_id):
    return stub.Get.with_call(
        environments_pb2.GetRequest(space_id=space_id, env_id=env_id),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def list(stub, space_id) -> tuple:
    return stub.List.with_call(
        environments_pb2.ListRequest(space_id=space_id),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def update(stub, space_id, env_id, description):
    return stub.Update.with_call(
        environments_pb2.UpdateRequest(env=environments_pb2.Environment(
            space_id=space_id,
            id=env_id,
            description=description
        )),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def set_alias(stub, space_id, env_id, alias):
    return stub.SetAlias.with_call(
        environments_pb2.SetAliasRequest(
            space_id=space_id,
            env_id=env_id,
            alias=alias
        ),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def remove_alias(stub, space_id, env_id, alias):
    return stub.SetAlias.with_call(
        environments_pb2.RemoveAliasRequest(
            space_id=space_id,
            env_id=env_id,
            alias=alias
        ),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def delete(stub, space_id, env_id) -> tuple:
    return stub.Delete.with_call(
        environments_pb2.DeleteRequest(space_id=space_id, env_id=env_id),
        metadata=(
            ('authorization', f'{ACCESS_TOKEN}'),
        )
    )


def main() -> None:
    with grpc.insecure_channel('envoy.perxis.pt.perx.ru:80') as channel:
        stub = environments_pb2_grpc.EnvironmentsStub(channel)
        print('---------- CREATE ----------')
        response, state = create(stub, SPACE_ID, 'test description')
        created = response.created
        print(response.created, state)
        print('---------- GET ----------')
        response, state = get(stub, SPACE_ID, created.id)
        print(response.env, state)
        print('---------- LIST ----------')
        response, state = list(stub, SPACE_ID)
        print(response.envs, state)
        print('---------- UPDATE ----------')
        response, state = update(stub, SPACE_ID, created.id, 'new test description')
        print(state)
        print('---------- SET ALIAS ----------')
        response, state = set_alias(stub, SPACE_ID, created.id, 'test_alias')
        print(state)
        print('---------- REMOVE ALIAS ----------')
        response, state = remove_alias(stub, SPACE_ID, created.id, 'test_alias')
        print(state)
        print('---------- DELETE ----------')
        response, state = delete(stub, SPACE_ID, created.id)
        print(state)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    main()