From 12ce257e6a71ecfec9624b75081bd6c1c55746bd Mon Sep 17 00:00:00 2001 From: Tomasz Durda Date: Sun, 21 Mar 2021 21:39:09 +0100 Subject: [PATCH 1/2] added qty to POST /printjobs, added pagination to GET /computers, GET /printers, GET /printjobs and GET /printjobs/states --- printnodeapi/computers.py | 67 +++++++++++++++++++++++++++++++++------ 1 file changed, 57 insertions(+), 10 deletions(-) diff --git a/printnodeapi/computers.py b/printnodeapi/computers.py index c5614e9..1ef67f3 100644 --- a/printnodeapi/computers.py +++ b/printnodeapi/computers.py @@ -11,17 +11,25 @@ def __init__(self, auth, factory): self._auth = auth self._factory = factory - def get_computers(self, computer=None): + 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): - results = self._auth.get('/computers') - computers = self._factory.create_computers(results) + 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) - results = self._auth.get('/computers/{}'.format(computer_id)) + 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)) @@ -41,15 +49,19 @@ def get_scales(self, computer, dev_name=None, dev_num=None): scales = self._factory.create_scales(self._auth.get(url)) return scales - def get_states(self, pjob_set=None): + 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("/printjobs/"+pjob_set+"states")) + self._auth.get(url)) return states - def get_printers(self, computer=None, printer=None): + 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 @@ -64,9 +76,13 @@ def get_printers(self, computer=None, printer=None): 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): @@ -75,17 +91,23 @@ def get_printers(self, computer=None, printer=None): return printers else: printer_id = self._get_printer_id(printer) - results = self._auth.get('/printers/{}'.format(printer_id)) - printers = self._factory.create_printers(results) + 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): + 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] @@ -103,6 +125,8 @@ def get_printjobs(self, computer=None, printer=None, printjob=None): 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): @@ -124,6 +148,7 @@ def submit_printjob( printer=None, job_type='pdf', # PDF|RAW title='PrintJob', + qty=None, options=None, authentication=None, uri=None, @@ -173,6 +198,9 @@ def submit_printjob( 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 @@ -255,6 +283,25 @@ def _get_printer_by_name(self, printer_name, computer_id=None): 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): From e4e8f85016dc6a3bdb862363356cc70d58be30e3 Mon Sep 17 00:00:00 2001 From: Tomasz Durda Date: Sun, 10 May 2026 19:31:16 +0200 Subject: [PATCH 2/2] fixes syntax warning in Python 3.12 and higher --- printnodeapi/auth.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/printnodeapi/auth.py b/printnodeapi/auth.py index 454ad71..4b4a710 100644 --- a/printnodeapi/auth.py +++ b/printnodeapi/auth.py @@ -119,7 +119,7 @@ def _request(self, method, endpoint, fields=None, request_headers=None): raise Exception('status code: ' + str(response.status_code)) def _is_hundreth(self, hundreth, number): - pattern = str(hundreth) + '\d{2,}' + pattern = str(hundreth) + r'\d{2,}' return bool(re.match(pattern, str(number))) def _fix_unicode(self, json_object):