pygbop 0.2.7.53

Geely GBOP Client

Geely GBOP Client

request demo:

from pygbop import GbopApiClient, Method, BasicAuth

auth = BasicAuth(access_key='xxxxxxxxxxxxxxxxx', secret_key='xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx')
client = GbopApiClient(auth, base_url='hello.api-dev.test.xxxxx.com')

print('=============GET=============')
params = {
    'params1': '123',
    'params3': ['s', 'w', 'k'],
    'params2': '321',
}
res = client.execute(Method.GET, '/api/v1/hello', params)
print(res.decode('utf-8'))

print('=============POST=============')
data = {'params3': 'testA', 'params4': 'testB'}
res = client.execute(Method.POST, '/api/v1/demo', data=data)
print(res.decode('utf-8'))

print('=============POST2=============')
params = {'params3': 'testA', 'params4': 'testB'}
header = {'content-type':'application/json;charset=utf-8'}
res = client.execute(Method.POST, '/api/v1/demo', params=params, data=data, header=header)
print(res.decode('utf-8'))

print('=============POST files=============')
file_paths = ['/path/to/file1.txt', '/path/to/file2.jpg']
files = [('file', (open(fp, 'rb'))) for fp in file_paths]
res = client.execute(Method.POST, '/api/v1/demo', files=files)
print(res.decode('utf-8'))
for _, file_obj in files:
    file_obj.close()


print('=============response stream=============')
for item in client.execute_with_stream(Method.POST, '/sse', params, data,
                                            header={"Content-Type": "application/json"},
                                            timeout=60):
    print(item.decode('utf-8'))

cloud event push demo:

from pygbop import CloudEventBasicAuth, EventPushClient

auth = CloudEventBasicAuth(producer_group='xxxxxxx',
                           subject='persistent://Information_Technology/xxxx/XXXX_EVENT',
                           secret_token='xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx')
event_push_client = EventPushClient(auth=auth, base_url='http://xxxx.dev.xxxxx.com')
data = {
    "message": "hello 喵啪斯"
}
response = event_push_client.push_message(data=data,
                                          source='xx.cmdb',
                                          type_='xx:cmdb:InstanceChanged')
print(response.decode('utf-8'))

or

from pygbop import CloudEventBasicAuth, EventPushClient

auth = CloudEventBasicAuth(producer_group='xxxxxxx',
                           subject='persistent://Information_Technology/xxxx/XXXX_EVENT',
                           secret_token='xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx')
event_push_client = EventPushClient(auth=auth, base_url='http://xxxx.dev.xxxxx.com')
data = {
    "message": "hello 喵啪斯"
}
response = event_push_client.push(data=data,
                                  source='xx.cmdb',
                                  type_='xx:cmdb:InstanceChanged')
if response.result:
    message_id = response.data
else:
    code = response.code
    fail_message = response.message