-
Notifications
You must be signed in to change notification settings - Fork 10
Expand file tree
/
Copy pathsemantics3.py
More file actions
131 lines (112 loc) · 4.44 KB
/
semantics3.py
File metadata and controls
131 lines (112 loc) · 4.44 KB
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import json
from requests_oauthlib import OAuth1Session
try:
import urllib.parse as urllib
except ImportError:
import urllib
try:
from .error import Semantics3Error
except ImportError:
from error import Semantics3Error
API_DOMAIN = 'api.semantics3.com'
API_BASE = 'https://' + API_DOMAIN + '/v1/'
class Semantics3Request:
def __init__(self, api_key=None, api_secret=None, endpoint=None):
if api_key is None:
raise Semantics3Error(
'API Credentials Missing',
'You did not supply an api_key. Please sign up at\
https://semantics3.com/ to obtain your api_key.'
)
if api_secret is None:
raise Semantics3Error(
'API Credentials Missing',
'You did not supply an api_secret. Please sign up at\
https://semantics3.com/ to obtain your api_key.'
)
self.api_key = api_key
self.api_secret = api_secret
self.endpoint = endpoint
self.oauth = OAuth1Session(api_key, client_secret=api_secret)
self.data_query = {}
self.query_result = None
self.cache_size = 10
def fetch(self, endpoint, params):
api_endpoint = API_BASE + endpoint + '?' +\
urllib.urlencode({'q': params})
content = self.oauth.get(api_endpoint,headers={'User-Agent':'Semantics3 Python Lib/0.2'})
return content
def remove(self, endpoint, *fields):
def _remove(path, hash):
if path[0] in hash:
if len(path) == 1:
del hash[path[0]]
else:
_remove(path[1:], hash[path[0]])
if not hash[path[0]]:
del hash[path[0]]
else:
raise Semantics3Error(
'Constraint does not exist',
"Constraint '" + path[0] + "' does not exist."
)
_remove(fields, self.data_query[endpoint])
def add(self, endpoint, *fields):
ancestors = fields[:-2]
parent = self.data_query.setdefault(endpoint, {})
for i in ancestors:
if isinstance(parent, dict):
parent = parent.setdefault(i, {})
else:
raise Semantics3Error(
'Invalid constraint',
'Cannot add this constraint, \'' + parent + '\' is already\
a value.'
)
if isinstance(parent, dict):
parent[fields[-2]] = fields[-1]
else:
raise Semantics3Error(
'Invalid constraint',
'Cannot add this constraint, \'' + parent + '\' is already\
a value.'
)
def field(self, *fields):
self.add(self.endpoint, *fields)
def cache(self, cache_size):
self.cache_size = cache_size
def iter(self):
self.add(self.endpoint, 'limit', self.cache_size)
self.run_query()
if('total_results_count' in self.query_result):
offset = 0
total_count = self.query_result['total_results_count']
while(offset < total_count):
for i in self.query_result['results']:
yield i
offset = offset + len(self.query_result['results'])
self.add(self.endpoint, 'offset', offset)
if(offset < total_count):
self.run_query()
def query(self, endpoint, **kwargs):
content = self.fetch(endpoint, json.dumps(kwargs)).content.decode('utf-8')
return json.loads(content)
def run_query(self, endpoint=None):
endpoint = endpoint or self.endpoint
if not endpoint in self.data_query:
raise Semantics3Error("No query built", "You need to first create\
a query using the add() method.")
query = self.data_query[endpoint]
self.query_result = self.query(
endpoint,
**query
)
if self.query_result['code'] != 'OK':
raise Semantics3Error(self.query_result['code'],
self.query_result['message'])
def get(self, endpoint=None):
self.run_query(endpoint)
return self.query_result
def clear_query(self):
self.data_query = {}
self.query_result = {}