dist/
MANIFEST
_build/
-doc/
+doc/_build
+doc/src.rst
+doc/src.cnfline.rst
+doc/src.cnfvar.rst
# PyDev
.project
--- /dev/null
+Overview Diagram of cnfvar package
+==================================
+
+::
+
+ frontend . CRUD layer . backend drivers . backend
+ | | |
+ | | | .-,( ),-.
+ o | | | .-( )-.
+ /|\ | .----------. | .---------->( varlink API )
+ / \ |/| CnfStore | | | | '-( ).-'
+ / '----------' | | | '-.( ).-'
+ /| | | .-------------. | |
+ .----------' | | | | varlink API | | |
+ | CnfList | | '---------------->| interface | | | _.-----._
+ '--.------. | | '-------------' | | .- -.
+ |.-----.\ | | | '->|-_ _-|
+ || Cnf | \| .----------------. | | | ~-----~ |
+ |'-----' \ | BinaryCnfStore | | | .->| cnfvar |
+ |.-----. |'-----------------' | | | `._backend_.'
+ || Cnf | | | | .------------. | | "-----"
+ |'-----' | | | | subprocess | | |
+ |.-----. | '-------------->| wrappers | | _______
+ '| Cnf | | | '------------' | |get_cnf|
+ '-----' | | | | |set_cnf|
+ | | '------------> |_______|
+ | | | /:::::::/
+ | | |
+ ' ' '
+
# General information about the project.
project = 'pyi2ncommon'
-copyright = '2016-2018, Intra2net AG <info@intra2net.com>'
+copyright = 'Intra2net AG <info@intra2net.com>'
author = 'Intra2net AG'
# The version info for the project you're documenting, acts as replacement for
# built documents.
#
# The short X.Y version.
-version = '0'
+version = '1.6'
# The full version, including alpha/beta/rc tags.
-release = '1'
+release = '7'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
# Example configuration for intersphinx: refer to the Python standard library.
-intersphinx_mapping = {'https://docs.python.org/3.3': None}
+intersphinx_mapping = {'https://docs.python.org/3': None}
# show doc of class AND of constructor
autoclass_content = 'class' # 'both' not needed if special-members is used
def skip_class_members(app, what, name, obj, skip, options):
exclude = name in ('__weakref__', '__dict__', '__module__', '__doc__') \
and what == 'class'
- if exclude:
- print('excluding {0}'.format(name))
+ #if exclude:
+ # print('excluding {0}'.format(name))
return skip or exclude
def setup(app):
license.rst
about_docu.rst
- template
+ cnfvar-api-ascii-art.rst
Modules
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)
from .binary import CnfBinary
from .store import CnfStore, BinaryCnfStore, CommitException
-__all__ = ["Cnf, CnfList, CnfBinary", "CnfStore",
+__all__ = ["Cnf", "CnfList", "CnfBinary", "CnfStore",
"BinaryCnfStore", "CommitException"]
"""
binary: wrappers around binaries that work with the CNF store.
-Featuring
-- CnfBinary: stateless class that can be used to invoke the different binaries
-in a Python-friendly way.
+Featuring:
+ - CnfBinary: stateless class that can be used to invoke the different binaries
+ in a Python-friendly way.
.. note:: It is written as a class on purpose for it to be easily extended to
-support invoking non-local binaries, but methods are all class methods since
-the class is stateless.
+ support invoking non-local binaries, but methods are all class methods since
+ the class is stateless.
+
+.. seealso:: Overview Diagram linked to from doc main page
.. codeauthor:: Intra2net
"""
import html
import re
-log = logging.getLogger("pyi2ncommon.cnfvar.store")
+log = logging.getLogger("pyi2ncommon.cnfvar.binary")
#: default get_cnf binary
BIN_GET_CNF = "/usr/intranator/bin/get_cnf"
ENCODING = "latin1"
-class CnfBinary(object):
+class CnfBinary:
"""Provide wrappers around the multiple binaries to handle the CNF store."""
@classmethod
:rtype: str
.. note:: being a wrapper, this function does not do anything extra
- like checking if arnied is running or waiting for generate.
+ like checking if arnied is running or waiting for generate.
"""
if name is None and instance is not None:
raise ValueError("cannot pass `instance` without a `name`")
:raises: :py:class:`SetCnfException` in case the binary errors out
.. note:: being a wrapper, this function does not do anything extra
- like checking if arnied is running or waiting for generate.
+ like checking if arnied is running or waiting for generate.
"""
if input_str is None and input_file is None:
raise ValueError("Need to pass either a string or a file")
"""
model: Cnf classes, collection of Cnf classes and multiple filtering methods.
-Featuring
-- Cnf: class representing a CNF variabe
-
-- CnfList: a collection of `Cnf` instances
+Featuring:
+ - Cnf: class representing a CNF variable
+ - CnfList: a collection of `Cnf` instances
The classes above inherit from their base types with added mixins which
extend them with extra functionality.
+.. seealso:: Overview Diagram linked to from doc main page
+
.. codeauthor:: Intra2net
"""
def __contains__(self, name):
return name.lower() in self.lower()
- def startswith(self, prefix):
- return self.lower().startswith(prefix.lower())
+ def startswith(self, prefix, *args, **kwargs):
+ return self.lower().startswith(prefix.lower(), *args, **kwargs)
- def endswith(self, prefix):
- return self.lower().endswith(prefix.lower())
+ def endswith(self, prefix, *args, **kwargs):
+ return self.lower().endswith(prefix.lower(), *args, **kwargs)
###############################################################################
else:
iter_ = []
super().__init__(iter_)
+ self._renumber_counter = None # initialized and used in renumber
if renumber:
self.renumber()
# NOTE: we don't keep track of operations that change the list as this
# would require us to reimplement most of the methods. At least for now
# this method should be called again when serializing.
- self._counter = 0
+ self._renumber_counter = 0
def renumber_fn(cnf):
- self._counter += 1
- cnf.lineno = self._counter
+ self._renumber_counter += 1
+ cnf.lineno = self._renumber_counter
self.for_each_all(renumber_fn)
:param where_filter: predicate to apply against CNFs
:type where_filter: function accepting a CNF and returning a boolean
- :returns: a instance of this class with filtered members
+ :returns: an instance of this class with filtered members
:rtype: :py:class:`CnfList`
"""
return CnfList(c for c in self if where_filter(c))
:param where_filter: predicate to apply against children
:type where_filter: function accepting a CNF and returning a boolean
- :returns: a instance of this class with filtered members
+ :returns: an instance of this class with filtered members
:rtype: :py:class:`CnfList`
"""
def upper_filter(cnf):
d["children"] = [_to_dict(c) for c in cnf.children]
return d
if renumber:
- renumber = True
+ self.renumber()
json_list = [_to_dict(c) for c in self]
return json.dumps({"cnf": json_list})
"""
cnf = self.children.first_with_name(name, default=None)
if cnf is None:
- cnf = self.add_child(name, "1")
+ self.add_child(name, "1")
else:
cnf.enable()
"""
cnf = self.children.first_with_name(name, default=None)
if cnf is None:
- cnf = self.add_child(name, "0")
+ self.add_child(name, "0")
else:
cnf.disable()
# Copyright (c) 2016-2022 Intra2net AG <info@intra2net.com>
"""
-store: implementations of CNF stores using varlink and *et_cnf binaries.
+store: implementations of CNF stores using varlink and `*et_cnf` binaries.
-Featuring
-- CnfStore: the main store class that is implemented using the varlink API
+Featuring:
+ - CnfStore: the main store class that is implemented using the varlink API
+ - BinaryCnfStore: alternative store class implementation using the `set_cnf`
+ and `get_cnf` binaries
-- BinaryCnfStore: alternative store class implementation using the set_ and
-get_cnf binaries
+.. seealso:: Overview Diagram linked to from doc main page
.. codeauthor:: Intra2net
"""
:param bool fix_problems: whether to automatically fix errors in the vars
.. note:: you can mix variables to insert and variables to update
- in the same list as the system should handle it nicely
+ in the same list as the system should handle it nicely
Example::
store = CnfStore()
class BinaryCnfStore(CnfStore):
- """Implementation of the CNF store that uses get_ and set_cnf."""
+ """Implementation of the CNF store that uses `get_cnf` and `set_cnf`."""
#: how much to wait for arnied to report running
ARNIED_TIMEOUT = 30
"""
log.debug("Querying BinaryCnfStore with name=%s and instance=%s",
name, instance)
- output = self._driver.get_cnf(name, instance)
+ output = self._driver.get_cnf(name, instance=instance)
if len(output) == 0:
# otherwise cnfvar raises a generic Malformed exception
:param bool fix_problems: whether to automatically fix errors in the vars
.. note:: you can mix variables to insert and variables to update
- in the same list as the system should handle it nicely
+ in the same list as the system should handle it nicely
Example::
store = CnfStore()
cnf = self._cnf_or_list(cnf, operation="commit")
self._autofix_instances(cnf)
- # set_cnf is demaning on lineno's
+ # set_cnf is demanding on lineno's
cnf.renumber()
log.debug("Committing variables via binaries:\n%s", cnf)
if any((c.parent is not None for c in cnf)):
raise RuntimeError("Calling delete is only supported on top-level CNF variables")
- # set_cnf is demaning on lineno's
+ # set_cnf is demanding on lineno's
cnf.renumber()
log.debug("Deleting variables via binaries:\n%s", cnf)
return fn(*args, **kwargs)
+#: pattern for more verbose error message for :py:class:`CommitException`
+COMMIT_EXCEPTION_MESSAGE = """\
+Error committing CNF variables!
+----------------------------
+Input:
+{cnfvars}
+----------------------------
+Error:
+{msg}
+"""
+
+
class CommitException(Exception):
"""Custom exception for commit errors."""
:type cnfvars: CnfList
:param str msg: error message
"""
- self.message = f"""\
-Error committing CNF variables!
-----------------------------
-Input:
-{cnfvars}
-----------------------------
-Error:
-{msg}
-"""
super().__init__(msg)
+ self.message = COMMIT_EXCEPTION_MESSAGE.format(cnfvars=cnfvars, msg=msg)
""" like the builtin print function but also accepts color args
If any arg of :py:func:`colored` is given in `kwargs`, will run text with
- color-args through that function. Runs built-in :py:builtin:`print`
+ color-args through that function. Runs built-in :py:func:`print`
function with result and other args.
...todo:: color all args, not just first
patch.object(store, "_wait_for_generate"):
self._query_and_update(store)
- cnf_api_input = set_cnf_mock.call_args.kwargs["vars"]
+ args, kwargs = set_cnf_mock.call_args
+ cnf_api_input = kwargs["vars"]
cnf_input = CnfList.from_api_structure(cnf_api_input)
expected_input = dedent("""\
patch.object(store, "_wait_for_generate"):
store.delete(cnfvars)
- cnf_api_input = set_cnf_mock.call_args.kwargs["vars"]
+ args, kwargs = set_cnf_mock.call_args
+ cnf_api_input = kwargs["vars"]
cnf_input = CnfList.from_api_structure(cnf_api_input)
expected_input = dedent("""\
git://developer.intra2net.com / pyi2ncommon / commitdiff