-
Notifications
You must be signed in to change notification settings - Fork 134
Expand file tree
/
Copy pathsearch.py
More file actions
142 lines (108 loc) · 4.2 KB
/
search.py
File metadata and controls
142 lines (108 loc) · 4.2 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
132
133
134
135
136
137
138
139
140
141
142
import json
import cloudinary
from cloudinary.api_client.call_api import call_json_api
from cloudinary.utils import (unique, build_distribution_domain, base64url_encode, json_encode, compute_hex_hash,
SIGNATURE_SHA256, build_array)
class Search(object):
ASSETS = 'resources'
_endpoint = ASSETS
_KEYS_WITH_UNIQUE_VALUES = {
'sort_by': lambda x: next(iter(x)),
'aggregate': lambda agg: agg["type"] if isinstance(agg, dict) and "type" in agg else agg,
'with_field': None,
'fields': None,
}
_ttl = 300 # Used for search URLs
"""Build and execute a search query."""
def __init__(self):
self.query = {}
def expression(self, value):
"""Specify the search query expression."""
self.query["expression"] = value
return self
def max_results(self, value):
"""Set the max results to return"""
self.query["max_results"] = value
return self
def next_cursor(self, value):
"""Get next page in the query using the ``next_cursor`` value from a previous invocation."""
self.query["next_cursor"] = value
return self
def sort_by(self, field_name, direction=None):
"""Add a field to sort results by. If not provided, direction is ``desc``."""
if direction is None:
direction = 'desc'
self._add("sort_by", {field_name: direction})
return self
def aggregate(self, value):
"""Aggregate field."""
self._add("aggregate", value)
return self
def with_field(self, value):
"""Request an additional field in the result set."""
self._add("with_field", value)
return self
def fields(self, value):
"""Request which fields to return in the result set."""
self._add("fields", value)
return self
def ttl(self, ttl):
"""
Sets the time to live of the search URL.
:param ttl: The time to live in seconds.
:return: self
"""
self._ttl = ttl
return self
def to_json(self):
return json.dumps(self.as_dict())
def execute(self, **options):
"""Execute the search and return results."""
options["content_type"] = 'application/json'
uri = [self._endpoint, 'search']
return call_json_api('post', uri, self.as_dict(), **options)
def as_dict(self):
to_return = {}
for key, value in self.query.items():
if key in self._KEYS_WITH_UNIQUE_VALUES:
value = unique(value, self._KEYS_WITH_UNIQUE_VALUES[key])
to_return[key] = value
return to_return
def to_url(self, ttl=None, next_cursor=None, **options):
"""
Creates a signed Search URL that can be used on the client side.
:param ttl: The time to live in seconds.
:param next_cursor: Starting position.
:param options: Additional url delivery options.
:return: The resulting search URL.
"""
api_secret = options.get("api_secret", cloudinary.config().api_secret or None)
if not api_secret:
raise ValueError("Must supply api_secret")
if ttl is None:
ttl = self._ttl
query = self.as_dict()
_next_cursor = query.pop("next_cursor", None)
if next_cursor is None:
next_cursor = _next_cursor
b64query = base64url_encode(json_encode(query, sort_keys=True))
prefix = build_distribution_domain(options)
signature = compute_hex_hash("{ttl}{b64query}{api_secret}".format(
ttl=ttl,
b64query=b64query,
api_secret=api_secret
), algorithm=SIGNATURE_SHA256)
return "{prefix}/search/{signature}/{ttl}/{b64query}{next_cursor}".format(
prefix=prefix,
signature=signature,
ttl=ttl,
b64query=b64query,
next_cursor="/{}".format(next_cursor) if next_cursor else "")
def endpoint(self, endpoint):
self._endpoint = endpoint
return self
def _add(self, name, value):
if name not in self.query:
self.query[name] = []
self.query[name].extend(build_array(value))
return self