Skip to content

Commit

Permalink
feat: add print and job handling methods to NodeAPI
Browse files Browse the repository at this point in the history
  • Loading branch information
joamag committed Jan 19, 2025
1 parent f1ee49e commit 9596ab2
Show file tree
Hide file tree
Showing 5 changed files with 162 additions and 16 deletions.
8 changes: 8 additions & 0 deletions src/colony_print/job.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-


class JobInfo(dict): ...


class JobsResult(dict): ...
12 changes: 12 additions & 0 deletions src/colony_print/job.pyi
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-

from typing import Mapping, TypedDict

class JobInfo(TypedDict):
name: str
node_id: str
printer: str | None
status: str | None

JobsResult = Mapping[str, JobInfo]
93 changes: 93 additions & 0 deletions src/colony_print/node.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,99 @@ def list_nodes(self, *args, **kwargs):
contents = self.get(url, **kwargs)
return contents

def jobs_node(self, id, *args, **kwargs):
url = self.base_url + "nodes/%s/jobs" % id
contents = self.get(url, **kwargs)
return contents

def print_default_node(
self,
id,
data=None,
data_b64=None,
name=None,
type=None,
format=None,
options=None,
*args,
**kwargs
):
url = self.base_url + "nodes/%s/print" % id
contents = self.post(
url,
data_j=dict(
data=data,
data_b64=data_b64,
name=name,
type=type,
format=format,
options=options,
),
**kwargs
)
return contents

def print_hello_default_node(
self, id, type=None, format=None, options=None, *args, **kwargs
):
url = self.base_url + "nodes/%s/print_hello" % id
contents = self.post(
url,
data_j=dict(
type=type,
format=format,
options=options,
),
**kwargs
)
return contents

def print_printer_node(
self,
id,
printer,
data=None,
data_b64=None,
name=None,
type=None,
format=None,
options=None,
*args,
**kwargs
):
url = self.base_url + "nodes/%s/printers/print" % id
contents = self.post(
url,
data_j=dict(
printer=printer,
data=data,
data_b64=data_b64,
name=name,
type=type,
format=format,
options=options,
),
**kwargs
)
return contents

def print_hello_printer_node(
self, id, printer, type=None, format=None, options=None, *args, **kwargs
):
url = self.base_url + "nodes/%s/printers/print_hello" % id
print(url)
contents = self.post(
url,
data_j=dict(
printer=printer,
type=type,
format=format,
options=options,
),
**kwargs
)
return contents


class Node(dict):
pass
39 changes: 39 additions & 0 deletions src/colony_print/node.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@

from typing import NotRequired, Sequence, TypedDict

from .job import JobsResult

class Node(TypedDict):
name: str
mode: str
Expand All @@ -12,3 +14,40 @@ class Node(TypedDict):

class NodeAPI:
def list_nodes(self) -> Sequence[Node]: ...
def jobs_node(self, id: str) -> JobsResult: ...
def print_default_node(
self,
id: str,
data: str | None = None,
data_b64: str | None = None,
name: str | None = None,
type: str | None = None,
format: str | None = None,
options: dict | None = None,
) -> str: ...
def print_hello_default_node(
self,
id: str,
type: str | None = None,
format: str | None = None,
options: dict | None = None,
) -> str: ...
def print_printer_node(
self,
id: str,
printer: str,
data: str | None = None,
data_b64: str | None = None,
name: str | None = None,
type: str | None = None,
format: str | None = None,
options: dict | None = None,
) -> str: ...
def print_hello_printer_node(
self,
id: str,
printer: str,
type: str | None = None,
format: str | None = None,
options: dict | None = None,
) -> str: ...
26 changes: 10 additions & 16 deletions src/colony_print/scripts/printer.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,28 +7,22 @@
import colony_print


def print(*args, **kwargs):
def print(id, printer=None):
api = colony_print.API()
return api.send(kwargs)
if printer:
return api.print_hello_printer_node(id, printer)
else:
return api.print_hello_default_node(id)


if __name__ == "__main__":
node = appier.conf("NODE", [], cast=list)
node = appier.conf("NODE", None)
printer = appier.conf("PRINTER", None)

kwargs = dict()
if receivers:
kwargs["receivers"] = receivers
if subject:
kwargs["subject"] = subject
if title:
kwargs["title"] = title
if contents:
kwargs["contents"] = contents
if copyright:
kwargs["copyright"] = copyright

result = send(**kwargs)
if not node:
raise appier.OperationalError(message="No node defined")

result = print(node, printer=printer)
pprint.pprint(result)
else:
__path__ = []

0 comments on commit 9596ab2

Please sign in to comment.