forked from PrintNode/PrintNode-Python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcomputers.py
More file actions
317 lines (279 loc) · 11.8 KB
/
Copy pathcomputers.py
File metadata and controls
317 lines (279 loc) · 11.8 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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
from printnodeapi.model import Computer, Model, Printer, PrintJob, Scale, State
from future.types import newbytes
import json
import base64 as base_64
import sys
class Computers:
def __init__(self, auth, factory):
self._auth = auth
self._factory = factory
def get_computers(self, computer=None, limit=None, after=None, dir=None):
params = self._create_pagination_params(limit, after, dir)
if self._is_multi_query(computer):
url = '/computers'
if params is not None:
url = url + '?' + params
computers = self._factory.create_computers(self._auth.get(url))
if isinstance(computer, str):
return [c for c in computers if c.name == computer]
else:
return computers
else:
computer_id = self._get_computer_id(computer)
url = '/computers/{}'.format(computer_id)
if params is not None:
url = url + '?' + params
results = self._auth.get(url)
if len(results) == 0:
raise LookupError(
'computer not found with ID {}'.format(computer_id))
assert len(results) == 1
computers = self._factory.create_computers(results)
return computers[0]
def get_scales(self, computer, dev_name=None, dev_num=None):
url = '/computer/'+str(computer)+'/scales'
if dev_name is not None:
url = url+'/'+dev_name
if dev_num is not None:
url = url+'/'+str(dev_num)
if dev_num is not None and dev_name is None:
temp_str = 'Device num stated without name - nothing found.'
raise LookupError(temp_str)
scales = self._factory.create_scales(self._auth.get(url))
return scales
def get_states(self, pjob_set=None, limit=None, after=None, dir=None):
pjob_set = str(pjob_set)+"/" if pjob_set else ""
url = "/printjobs/"+pjob_set+"states"
params = self._create_pagination_params(limit, after, dir)
if params is not None:
url = url + '?' + params
states = self._factory.create_states_map(
self._auth.get(url))
return states
def get_printers(self, computer=None, printer=None, limit=None, after=None, dir=None):
"""queries API for printers.
the printer argument can be:
* id of the printer, in which case a single printer is returned
* name of the printer, in which case a single printer is returned
* unspecified in which case a list of all printers is returned
the computer argument can be:
* id of the computer in which case only printers that are attached
to that computer are returned
* unspecified in which case all printers attached to the account
are considered.
raises LookupError if querying for a specific printer
that doesn't exist
"""
params = self._create_pagination_params(limit, after, dir)
if self._is_multi_query(printer):
computer_ids = ','.join(map(str, self._get_computer_ids(computer)))
url = '/computers/{}/printers'.format(computer_ids)
if params is not None:
url = url + '?' + params
printers = self._factory.create_printers(self._auth.get(url))
assert all(isinstance(p, Printer) for p in printers)
if isinstance(printer, str):
return [p for p in printers if p.name == printer]
else:
return printers
else:
printer_id = self._get_printer_id(printer)
url = '/printers/{}'.format(printer_id)
if params is not None:
url = url + '?' + params
printers = self._factory.create_printers(self._auth.get(url))
assert all(isinstance(p, Printer) for p in printers)
if len(printers) == 0:
raise LookupError('no printer with ID {}'.format(printer_id))
return printers[0]
def get_printjobs(self, computer=None, printer=None, printjob=None, limit=None, after=None, dir=None):
params = self._create_pagination_params(limit, after, dir)
if self._is_multi_query(printjob):
if computer is None and printer is None:
url = '/printjobs'
if params is not None:
url = url + '?' + params
printjobs = self._factory.create_printjobs(self._auth.get(url))
if isinstance(printjob, str):
return [pj for pj in printjobs if pj.title == printjob]
else:
return printjobs
else:
printers = self.get_printers(
computer=computer,
printer=printer)
if self._is_multi_query(printer):
if len(printers) == 0:
return []
else:
printers = [printers]
printer_ids = [p.id for p in printers]
printer_url = ','.join(map(str, printer_ids))
url = '/printers/{}/printjobs'.format(printer_url)
if params is not None:
url = url + '?' + params
printjobs = self._factory.create_printjobs(self._auth.get(url))
if isinstance(printjob, str):
return [pj for pj in printjobs if pj.title == printjob]
else:
return printjobs
else:
printjob_id = self._get_printjob_id(printjob)
results = self._auth.get('/printjobs/{}'.format(printjob_id))
printjobs = self._factory.create_printjobs(results)
if len(printjobs) == 0:
raise LookupError('no printjob with ID {}'.format(printjob_id))
return printjobs[0]
# could use the default printer if none is provided
def submit_printjob(
self,
computer=None,
printer=None,
job_type='pdf', # PDF|RAW
title='PrintJob',
qty=None,
options=None,
authentication=None,
uri=None,
base64=None,
binary=None):
printers = self.get_printers(computer=computer, printer=printer)
assert isinstance(printers, (list, Printer))
if isinstance(printers, list):
if len(printers) == 0:
raise LookupError('printer not found')
elif len(printers) > 1:
printer_ids = ','.join(p.id for p in printers)
msg = 'multiple printers match destination: {}'.format(
printer_ids)
raise LookupError(msg)
printer_id = printers[0].id
else:
printer_id = printers.id
if job_type not in ['pdf', 'raw', 'binary']:
raise ValueError('only support job_type of pdf or raw')
if len([x for x in [uri, base64, binary] if x is not None]) != 1:
raise ValueError('one and only one of the following parameters '
'is needed: uri, base64, or binary')
if binary is not None:
if sys.version_info[0] < 3:
binary_bytes = newbytes(binary)
else:
if isinstance(binary, str):
binary_bytes = binary.encode('latin-1')
else:
binary_bytes = binary
base64 = base_64.b64encode(binary_bytes)
base64 = base64.decode('utf-8')
printjob_data = {
'printerId': printer_id,
'title': title,
'contentType': job_type + '_' + ('uri' if uri else 'base64'),
'content': uri or base64,
'source': 'PythonApiClient'}
if authentication is not None:
printjob_data.update({"authentication": authentication})
if options is not None:
printjob_data.update({"options": options})
if qty is not None:
printjob_data.update({"qty": qty})
printjob_id = self._auth.post('/printjobs', printjob_data)
return printjob_id
def _native(self, obj):
if hasattr(obj, '__native__'):
return obj.__native__()
else:
return obj
def _get_computer_ids(self, computer):
if isinstance(computer, int):
return [computer]
elif isinstance(computer, Computer):
return [computer.id]
elif computer is None or isinstance(computer, str):
computers = [
self._factory.create_computer(comp)
for comp in self._auth.get('/computers')]
if isinstance(computer, str):
computers = [
comp
for comp in computers
if comp.name == computer]
return [c.id for c in computers]
else:
raise TypeError('computer: "{}"'.format(type(computer)))
def _get_computer_id(self, computer):
return self._get_model_id(computer, Computer)
def _get_printer_id(self, printer):
return self._get_model_id(printer, Printer)
def _get_printjob_id(self, printjob):
return self._get_model_id(printjob, PrintJob)
def _get_model_id(self, model, model_type):
if isinstance(model, int):
return model
elif isinstance(model, model_type):
return model.id
else:
raise TypeError(str(type(model)))
def _get_printer_ids(self, printer):
if isinstance(printer, int):
return [printer]
elif isinstance(printer, Printer):
return [Printer.id]
elif printer is None or isinstance(printer, str):
printers = {
self._factory.create_printer(printer_dict)
for printer_dict in self._auth.get('/printer')}
if isinstance(printer, str):
printers = {
p for p in printers if p.name == printer}
return printers
else:
raise TypeError('printer: "{}"'.format(type(printer)))
def _is_multi_query(self, obj):
if obj is None:
return True
elif isinstance(obj, int):
return False
elif isinstance(obj, str):
return True
elif isinstance(obj, Model):
return False
else:
raise TypeError('type "{}" unsupported'.format(type(obj)))
def _get_printer_by_name(self, printer_name, computer_id=None):
assert isinstance(printer_name, str)
assert not computer_id or isinstance(computer_id, int)
printers = self.get_printers(computer=computer_id)
return [
printer
for printer in printers
if printer.name == printer_name]
def _create_pagination_params(self, limit, after, dir):
params = []
if limit is not None:
if isinstance(limit, int):
params.append('limit={}'.format(limit))
else:
raise TypeError('limit: "{}"'.format(type(limit)))
if after is not None:
if isinstance(after, int):
params.append('after={}'.format(after))
else:
raise TypeError('after: "{}"'.format(type(after)))
if dir is not None:
if dir == 'asc' or dir == 'desc':
params.append('dir={}'.format(dir))
else:
raise TypeError("dir must equal 'asc' or 'desc'")
if len(params) > 0:
return '&'.join(params)
class LookupFailedError(RuntimeError):
def __init__(self, obj_name, field, value):
msg = 'Failed to find a matching {} with {} of {}'.format(
obj_name,
field,
value)
super(RuntimeError, self).__init__(msg)
self.obj_name = obj_name
self.field = field
self.value = value