-
Notifications
You must be signed in to change notification settings - Fork 0
/
dwb.py
78 lines (63 loc) · 2.56 KB
/
dwb.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# -*- coding: utf-8 -*-
#Copyright (c) 2015 - Ian <dengxw@eship.com.cn>
"""
#python wrapper for adobe data workbench api
#function:
__init__(self,server,profile)
createRequest(self,query):
resultRequest(self,queryId,completion=0,format='json')
dropRequest(self,queryId):
getSchema(self,format='text'):
#value option:
server:
insight server,example:"127.0.0.1:81"
profile:
insight profile name,example:"My Profile"
query:
query expression,example:"eval Visits over Day;"
format:
data return format,including json,text and xml
completion:
query completion(sample rate)
value range from 0.0-1.0
#Usage:
import dwb
#init client
client=dwb.DwbClient('192.168.16.222:81','Plateno Prd')
#create query and get queryid
queryId=client.createRequest('eval Visits over Day;')
#get query result
result=client.resultRequest(queryId)
"""
import requests
import urllib
import logging
logging.basicConfig(filename='dwb_api.log',level=logging.WARNING,format='%(asctime)s %(message)s')
class DwbClient:
def __init__(self,server,profile):
self.url='http://'+server+'/Profiles/'+urllib.quote(profile)+'/API.query'
self.cHeaders={"X-Action":"create","X-Language":"Expression"}
self.session = requests.Session()
def createRequest(self,query):
r=self.session.post(self.url,headers=self.cHeaders,data=query)
return r.headers['X-Query-ID'] if r.status_code==200 else None
def resultRequest(self,queryId,completion=0,format='json'):
self.qHeaders={"X-Action":"result","X-Format":format,"X-Language":"Expression","X-Completion":completion,"X-Query-ID":queryId}
r=self.session.post(self.url,headers=self.qHeaders)
if r.status_code!=200:
logging.warning('Create Query Failed:'+str(r.status_code)+str(r.headers))
return r.content if r.status_code==200 else None
def dropRequest(self,queryId):
self.qHeaders={"X-Action":"drop","X-Query-ID":queryId}
r=self.session.post(self.url,headers=self.qHeaders)
if r.status_code!=200:
logging.warning('Drop Request Failed:'+str(r.status_code)+str(r.headers))
return True if r.status_code==200 else False
def getSchema(self,format='text'):
self.sHeaders={"X-Action":"get-schema","X-Format":format}
r=self.session.post(self.url,headers=self.sHeaders)
if r.status_code!=200:
logging.warning('Schema Fetch Failed:'+str(r.status_code)+str(r.headers))
return r.content if r.status_code==200 else None
def close(self):
self.session.close()