feat: add JobAPI and related functionality for job management

src/colony_print/__init__.py

@@ -2,7 +2,9 @@
 # -*- coding: utf-8 -*-
 
 from . import base
+from . import job
 from . import node
 
 from .base import BASE_URL, API, Ping
+from .job import JobAPI, JobInfo, JobsResult, PrintResult
 from .node import NodeAPI, Node

src/colony_print/base.py

@@ -3,14 +3,15 @@
 
 import appier
 
+from .job import JobAPI
 from .node import NodeAPI
 
 BASE_URL = "https://print.bemisc.com/api/"
 """ The default base URL to be used when no other
 base URL value is provided to the constructor """
 
 
-class API(appier.API, NodeAPI):
+class API(appier.API, JobAPI, NodeAPI):
     """
     Implementation of the Colony Print API specification
     for a simplified python client usage.

src/colony_print/base.pyi

@@ -1,12 +1,13 @@
 from typing import TypedDict
 from appier import API as BaseAPI
 
+from .job import JobAPI
 from .node import NodeAPI
 
 BASE_URL: str = ...
 
 class Ping(TypedDict):
     time: float
 
-class API(BaseAPI, NodeAPI):
+class API(BaseAPI, JobAPI, NodeAPI):
     def ping(self) -> Ping: ...

src/colony_print/job.py

@@ -2,7 +2,26 @@
 # -*- coding: utf-8 -*-
 
 
-class JobInfo(dict): ...
+class JobAPI(object):
 
+    def list_jobs(self, *args, **kwargs):
+        url = self.base_url + "jobs"
+        contents = self.get(url, **kwargs)
+        return contents
 
-class JobsResult(dict): ...
+    def get_job(self, id):
+        url = self.base_url + "jobs/%s" % id
+        contents = self.get(url)
+        return contents
+
+
+class JobInfo(dict):
+    pass
+
+
+class JobsResult(dict):
+    pass
+
+
+class PrintResult(dict):
+    pass

src/colony_print/job.pyi

@@ -1,12 +1,30 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
 
-from typing import Mapping, TypedDict
+from typing import Mapping, NotRequired, TypedDict
+
+class JobResult(TypedDict):
+    handler: str
+    result: str
+    receivers: NotRequired[list[str]]
+    output_data: NotRequired[str]
 
 class JobInfo(TypedDict):
     name: str
     node_id: str
     printer: str | None
     status: str | None
+    result: NotRequired[JobResult]
 
 JobsResult = Mapping[str, JobInfo]
+
+class PrintResult(TypedDict):
+    id: str
+    name: str
+    node_id: str
+    printer: str | None
+    data_length: int
+
+class JobAPI:
+    def list_jobs(self) -> JobsResult: ...
+    def get_job(self, id: str) -> JobInfo: ...

src/colony_print/node.py

@@ -1,6 +1,8 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
 
+import json
+
 
 class NodeAPI(object):
 
@@ -27,13 +29,13 @@ def print_default_node(
         url = self.base_url + "nodes/%s/print" % id
         contents = self.post(
             url,
-            data_j=dict(
+            params=dict(
                 data=data,
                 data_b64=data_b64,
                 name=name,
                 type=type,
                 format=format,
-                options=options,
+                options=json.dumps(options) if options else None,
             ),
         )
         return contents
@@ -48,10 +50,10 @@ def print_hello_default_node(
         url = self.base_url + "nodes/%s/print_hello" % id
         contents = self.post(
             url,
-            data_j=dict(
+            params=dict(
                 type=type,
                 format=format,
-                options=options,
+                options=json.dumps(options) if options else None,
             ),
         )
         return contents
@@ -70,14 +72,14 @@ def print_printer_node(
         url = self.base_url + "nodes/%s/printers/print" % id
         contents = self.post(
             url,
-            params=dict(printer=printer),
-            data_j=dict(
+            params=dict(
+                printer=printer,
                 data=data,
                 data_b64=data_b64,
                 name=name,
                 type=type,
                 format=format,
-                options=options,
+                options=json.dumps(options) if options else None,
             ),
         )
         return contents
@@ -86,14 +88,13 @@ def print_hello_printer_node(
         self, id, printer, type=None, format=None, options=None
     ):
         url = self.base_url + "nodes/%s/printers/print_hello" % id
-        print(printer)
         contents = self.post(
             url,
-            params=dict(printer=printer),
-            data_j=dict(
+            params=dict(
+                printer=printer,
                 type=type,
                 format=format,
-                options=options,
+                options=json.dumps(options) if options else None,
             ),
         )
         return contents

src/colony_print/node.pyi

@@ -3,7 +3,7 @@
 
 from typing import NotRequired, Sequence, TypedDict
 
-from .job import JobsResult
+from .job import JobsResult, PrintResult
 
 class Node(TypedDict):
     name: str
@@ -24,14 +24,14 @@ class NodeAPI:
         type: str | None = None,
         format: str | None = None,
         options: dict | None = None,
-    ) -> str: ...
+    ) -> PrintResult: ...
     def print_hello_default_node(
         self,
         id: str,
         type: str | None = None,
         format: str | None = None,
         options: dict | None = None,
-    ) -> str: ...
+    ) -> PrintResult: ...
     def print_printer_node(
         self,
         id: str,
@@ -42,12 +42,12 @@ class NodeAPI:
         type: str | None = None,
         format: str | None = None,
         options: dict | None = None,
-    ) -> str: ...
+    ) -> PrintResult: ...
     def print_hello_printer_node(
         self,
         id: str,
         printer: str,
         type: str | None = None,
         format: str | None = None,
         options: dict | None = None,
-    ) -> str: ...
+    ) -> PrintResult: ...

src/colony_print/scripts/__init__.py

@@ -3,4 +3,4 @@
 
 from . import printer
 
-from .printer import print
+from .printer import print, save

src/colony_print/scripts/printer.py

@@ -1,6 +1,8 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
 
+import time
+import base64
 import pprint
 
 import appier
@@ -15,14 +17,46 @@ def print(id, printer=None):
         return api.print_hello_default_node(id)
 
 
+def save(id, printer=None):
+    api = colony_print.API()
+    options = dict(send_email=False, save_output=True)
+    if printer:
+        result = api.print_hello_printer_node(id, printer, options=options)
+    else:
+        result = api.print_hello_default_node(id, options=options)
+
+    job_id = result["id"]
+
+    iterations = 0
+    while iterations < 10:
+        time.sleep(0.5)
+        try:
+            job = api.get_job(job_id)
+        except Exception as error:
+            continue
+        if not "result" in job:
+            continue
+        result = job["result"]
+        if not "output_data" in result:
+            continue
+        data_b64 = result["output_data"]
+        data = base64.b64decode(data_b64)
+        with open("output.pdf", "wb") as file:
+            file.write(data)
+        break
+
+    return result
+
+
 if __name__ == "__main__":
     node = appier.conf("NODE", None)
     printer = appier.conf("PRINTER", None)
+    save_pdf = appier.conf("SAVE_PDF", False, cast=bool)
 
     if not node:
         raise appier.OperationalError(message="No node defined")
 
-    result = print(node, printer=printer)
+    result = save(node, printer=printer) if save_pdf else print(node, printer=printer)
     pprint.pprint(result)
 else:
     __path__ = []