Featuring
- Arnied: stateless class with methods as exposed in the varlink API.
+For documentation of Exceptions, methods and their arguments, refer to arnied
+source code (arnied/client/arnieclient.hxx). For compatibility, argument names
+are the same here as they are there (defeating python naming rules).
+
+See package `cnfvar` for a higher-level interface to this functionality.
+
.. codeauthor:: Intra2net
"""
@dataclass
-class CnfVar(object):
+class CnfVar:
name: str
instance: int
data: str
deleted: bool
children: typing.List['CnfVar']
+
@dataclass
-class ChangeCnfVar(object):
+class ChangeCnfVar:
chg_nr: int
name: str
instance: int
result_type: int
result_msg: str
+
@dataclass
-class GetCnfQuery(object):
+class GetCnfQuery:
name: str
instance: typing.Optional[int] = None
+
class ProgramStatus(Enum):
Running = auto()
Scheduled = auto()
class IsQueueActiveRet(SimpleNamespace):
active: bool
+
class IsScheduledOrRunningRet(SimpleNamespace):
status: ProgramStatus
timestamp: typing.Optional[int]
+
class GetCnfRet(SimpleNamespace):
vars: typing.List[CnfVar]
+
class SetCnfRet(SimpleNamespace):
change_numbers: typing.List[int]
+
class SetCommitCnf(SimpleNamespace):
results: typing.List[ChangeCnfVar]
+
class CommitCnfRet(SimpleNamespace):
results: typing.List[ChangeCnfVar]
def __init__(self):
super().__init__("An error occurred (InternalBridgeError)")
+
class EmptyInputError(Exception):
def __init__(self):
super().__init__("An error occurred (EmptyInputError)")
+
class BadCharError(Exception):
def __init__(self, results: str) -> None:
super().__init__(f"[BadCharError] Error in the arnied API (results={results})")
self.results = results
+
class ChildError(Exception):
def __init__(self, results: str) -> None:
super().__init__(f"[ChildError] Error in the arnied API (results={results})")
self.results = results
+
class CnfCommitError(Exception):
def __init__(self, results: typing.List[ChangeCnfVar]) -> None:
self.results = []
r = ChangeCnfVar(**r)
self.results.append(r)
msgs.append(f"{r.name},{r.instance}: \"{r.data}\": {r.result_msg}")
- super().__init__("Error commiting cnfvars:\n" + "\n".join(msgs))
+ super().__init__("Error committing cnfvars:\n" + "\n".join(msgs))
+
class NotFoundError(Exception):
def __init__(self, results: str) -> None:
super().__init__(f"[NotFoundError] Error in the arnied API (results={results})")
self.results = results
+
class SchedulerProgError(Exception):
def __init__(self):
super().__init__("An error occurred (SchedulerProgError)")
+
class ShowerrVarDataErr(Exception):
def __init__(self, msg_id: int) -> None:
super().__init__(f"[ShowerrVarDataErr] Error in the arnied API (msg_id={msg_id})")
self.msg_id = msg_id
+
class ShowerrVarMissing(Exception):
def __init__(self, params_count: int) -> None:
- super().__init__(f"[ShowerrVarMissing] Error in the arnied API (params_count={params_count})")
+ super().__init__(
+ f"[ShowerrVarMissing] Error in the arnied API (params_count={params_count})")
self.params_count = params_count
+
class ProviderNotFound(Exception):
def __init__(self, provider_id: int) -> None:
- super().__init__(f"[ProviderNotFound] Not found a provider with ID `{provider_id})`")
+ super().__init__(
+ f"[ProviderNotFound] Could not find provider (provider_id={provider_id})")
self.provider_id = provider_id
# Arnied varlink proxy
-class Arnied(object):
+class Arnied:
+ """
+ Expose methods of the arnied varlink interface in python.
+
+ As described in module doc, documentation of exceptions, methods, and
+ their arguments can be found in arnied source code.
+ """
VARLINK_ADDRESS = f"unix:{ARNIED_SOCKET}"
@classmethod
return conn.SetCnf(vars)
@classmethod
- def set_commit_cnf(cls, vars: typing.List[CnfVar], username: typing.Optional[str], nogenerate: bool, fix_commit: bool) -> SetCommitCnf:
+ def set_commit_cnf(cls, vars: typing.List[CnfVar], username: typing.Optional[str],
+ nogenerate: bool, fix_commit: bool) -> SetCommitCnf:
with cls.new_connection() as conn:
return conn.SetCommitCnf(vars, username, nogenerate, fix_commit)
@classmethod
- def commit_cnf(cls, change_numbers: typing.List[int], username: typing.Optional[str], nogenerate: bool, fix_commit: bool) -> CommitCnfRet:
+ def commit_cnf(cls, change_numbers: typing.List[int], username: typing.Optional[str],
+ nogenerate: bool, fix_commit: bool) -> CommitCnfRet:
with cls.new_connection() as conn:
return conn.CommitCnf(change_numbers, username, nogenerate, fix_commit)