Merge pull request #124 from nolar/renames-and-moves

Move the modules around to cleanup the code

kopf/__init__.py

@@ -9,8 +9,11 @@
 from kopf import (
     on,  # as a separate name on the public namespace
 )
-from kopf.config import (
+from kopf.clients.auth import (
     login,
+    LoginError,
+)
+from kopf.config import (
     configure,
     LOGLEVEL_INFO,
     LOGLEVEL_WARNING,
@@ -65,7 +68,8 @@
 
 __all__ = [
     'on', 'lifecycles', 'register', 'execute',
-    'login', 'configure',
+    'configure',
+    'login', 'LoginError',
     'event', 'info', 'warn', 'exception',
     'run', 'create_tasks',
     'adopt', 'label',

kopf/cli.py

@@ -4,9 +4,17 @@
 import click
 
 from kopf import config
-from kopf.reactor import loading
-from kopf.reactor import peering
+from kopf.clients import auth
+from kopf.engines import peering
 from kopf.reactor import queueing
+from kopf.utilities import loaders
+
+
+def cli_login():
+    try:
+        auth.login()
+    except auth.LoginError as e:
+        raise click.ClickException(str(e))
 
 
 def logging_options(fn):
@@ -39,8 +47,8 @@ def main():
 @click.argument('paths', nargs=-1)
 def run(paths, modules, peering_name, priority, standalone, namespace):
     """ Start an operator process and handle all the requests. """
-    config.login()
-    loading.preload(
+    cli_login()
+    loaders.preload(
         paths=paths,
         modules=modules,
     )
@@ -63,7 +71,7 @@ def run(paths, modules, peering_name, priority, standalone, namespace):
 @click.option('-m', '--message', type=str)
 def freeze(id, message, lifetime, namespace, peering_name, priority):
     """ Freeze the resource handling in the cluster. """
-    config.login()
+    cli_login()
     ourserlves = peering.Peer(
         id=id or peering.detect_own_id(),
         name=peering_name,
@@ -82,7 +90,7 @@ def freeze(id, message, lifetime, namespace, peering_name, priority):
 @click.option('-P', '--peering', 'peering_name', type=str, default=None, envvar='KOPF_RESUME_PEERING')
 def resume(id, namespace, peering_name):
     """ Resume the resource handling in the cluster. """
-    config.login()
+    cli_login()
     ourselves = peering.Peer(
         id=id or peering.detect_own_id(),
         name=peering_name,

kopf/k8s/__init__.py → kopf/clients/__init__.py

@@ -1,10 +1,10 @@
 """
-All the routines to talk to Kubernetes API.
+All the routines to talk to Kubernetes API and other APIs.
 
 This library is supposed to be mocked when the mocked K8s client is needed,
 and only the high-level logic has to be tested, not the API calls themselves.
 
-Beware: this is NOT a Kubernetes client. It is set of dedicated helpers
+Beware: this is NOT a Kubernetes client. It is set of dedicated adapters
 specially tailored to do the framework-specific tasks, not the generic
 Kubernetes object manipulation.
 

kopf/clients/auth.py

@@ -0,0 +1,42 @@
+import logging
+
+import kubernetes
+import urllib3.exceptions
+
+logger = logging.getLogger(__name__)
+
+
+class LoginError(Exception):
+    """ Raised when the operator cannot login to the API. """
+
+
+def login():
+    """
+    Login the the Kubernetes cluster, locally or remotely.
+    """
+
+    # Configure the default client credentials for all possible environments.
+    try:
+        kubernetes.config.load_incluster_config()  # cluster env vars
+        logger.debug("configured in cluster with service account")
+    except kubernetes.config.ConfigException as e1:
+        try:
+            kubernetes.config.load_kube_config()  # developer's config files
+            logger.debug("configured via kubeconfig file")
+        except kubernetes.config.ConfigException as e2:
+            raise LoginError(f"Cannot authenticate neither in-cluster, nor via kubeconfig.")
+
+    # Make a sample API call to ensure the login is successful,
+    # and convert some of the known exceptions to the CLI hints.
+    try:
+        api = kubernetes.client.CoreApi()
+        api.get_api_versions()
+    except urllib3.exceptions.HTTPError as e:
+        raise LoginError("Cannot connect to the Kubernetes API. "
+                         "Please configure the cluster access.")
+    except kubernetes.client.rest.ApiException as e:
+        if e.status == 401:
+            raise LoginError("Cannot authenticate to the Kubernetes API. "
+                             "Please login or configure the tokens.")
+        else:
+            raise

kopf/k8s/watching.py → kopf/clients/watching.py

@@ -24,7 +24,7 @@
 
 import kubernetes
 
-from kopf.k8s import fetching
+from kopf.clients import fetching
 from kopf.reactor import registries
 
 logger = logging.getLogger(__name__)

kopf/config.py

@@ -4,12 +4,9 @@
 import logging
 from typing import Optional
 
-import click
 import kubernetes
 import kubernetes.client.rest
-import urllib3.exceptions
 
-logger = logging.getLogger(__name__)
 format = '[%(asctime)s] %(name)-20.20s [%(levelname)-8.8s] %(message)s'
 
 
@@ -23,42 +20,6 @@
 """ Event loglevel to log only critical events(basically - no events). """
 
 
-class LoginError(click.ClickException):
-    """ Raised when the operator cannot login to the API. """
-
-
-def login():
-    """
-    Login the the Kubernetes cluster, locally or remotely.
-    """
-
-    # Configure the default client credentials for all possible environments.
-    try:
-        kubernetes.config.load_incluster_config()  # cluster env vars
-        logger.debug("configured in cluster with service account")
-    except kubernetes.config.ConfigException as e1:
-        try:
-            kubernetes.config.load_kube_config()  # developer's config files
-            logger.debug("configured via kubeconfig file")
-        except kubernetes.config.ConfigException as e2:
-            raise LoginError(f"Cannot authenticate neither in-cluster, nor via kubeconfig.")
-
-    # Make a sample API call to ensure the login is successful,
-    # and convert some of the known exceptions to the CLI hints.
-    try:
-        api = kubernetes.client.CoreApi()
-        api.get_api_versions()
-    except urllib3.exceptions.HTTPError as e:
-        raise LoginError("Cannot connect to the Kubernetes API. "
-                         "Please configure the cluster access.")
-    except kubernetes.client.rest.ApiException as e:
-        if e.status == 401:
-            raise LoginError("Cannot authenticate to the Kubernetes API. "
-                             "Please login or configure the tokens.")
-        else:
-            raise
-
-
 def configure(debug=None, verbose=None, quiet=None):
     log_level = 'DEBUG' if debug or verbose else 'WARNING' if quiet else 'INFO'
 

kopf/engines/__init__.py

@@ -0,0 +1,8 @@
+"""
+Engines are things that run around the reactor (see `kopf.reactor`)
+to help it to function at full strength, but are not part of it.
+For example, all never-ending side-tasks for peering and k8s-event-posting.
+
+The reactor and engines exchange the state with each other (bi-directionally)
+via the provided synchronization objects, usually asyncio events & queues.
+"""

kopf/reactor/peering.py → kopf/engines/peering.py

@@ -40,8 +40,8 @@
 
 import iso8601
 
-from kopf.k8s import fetching
-from kopf.k8s import patching
+from kopf.clients import fetching
+from kopf.clients import patching
 from kopf.reactor import registries
 
 logger = logging.getLogger(__name__)

kopf/events.py

@@ -15,7 +15,7 @@
 import sys
 
 from kopf import config
-from kopf.k8s import events
+from kopf.clients import events
 
 
 # TODO: rename it it kopf.log()? kopf.events.log()? kopf.events.warn()?

kopf/reactor/handling.py

@@ -22,7 +22,7 @@
 from typing import Optional, Callable, Iterable, Union, Collection
 
 from kopf import events
-from kopf.k8s import patching
+from kopf.clients import patching
 from kopf.reactor import causation
 from kopf.reactor import invocation
 from kopf.reactor import registries

kopf/reactor/queueing.py

@@ -32,10 +32,10 @@
 import aiojobs
 
 from kopf import config
-from kopf.k8s import watching
+from kopf.clients import watching
+from kopf.engines import peering
 from kopf.reactor import handling
 from kopf.reactor import lifecycles
-from kopf.reactor import peering
 from kopf.reactor import registries
 
 logger = logging.getLogger(__name__)
@@ -157,6 +157,7 @@ async def worker(
 
 
 def create_tasks(
+        loop: asyncio.AbstractEventLoop,
         lifecycle: Optional[Callable] = None,
         registry: Optional[registries.BaseRegistry] = None,
         standalone: bool = False,
@@ -183,35 +184,38 @@ def create_tasks(
     )
     if ourselves:
         tasks.extend([
-            asyncio.Task(peering.peers_keepalive(ourselves=ourselves)),
-            asyncio.Task(watcher(namespace=namespace,
-                                 resource=ourselves.resource,
-                                 handler=functools.partial(peering.peers_handler,
-                                                           ourselves=ourselves,
-                                                           freeze=freeze))),  # freeze is set/cleared
+            loop.create_task(peering.peers_keepalive(
+                ourselves=ourselves)),
+            loop.create_task(watcher(
+                namespace=namespace,
+                resource=ourselves.resource,
+                handler=functools.partial(peering.peers_handler,
+                                          ourselves=ourselves,
+                                          freeze=freeze))),  # freeze is set/cleared
         ])
 
     # Resource event handling, only once for every known resource (de-duplicated).
     for resource in registry.resources:
         tasks.extend([
-            asyncio.Task(watcher(namespace=namespace,
-                                 resource=resource,
-                                 handler=functools.partial(handling.custom_object_handler,
-                                                           lifecycle=lifecycle,
-                                                           registry=registry,
-                                                           resource=resource,
-                                                           freeze=freeze))),  # freeze is only checked
+            loop.create_task(watcher(
+                namespace=namespace,
+                resource=resource,
+                handler=functools.partial(handling.custom_object_handler,
+                                          lifecycle=lifecycle,
+                                          registry=registry,
+                                          resource=resource,
+                                          freeze=freeze))),  # freeze is only checked
         ])
 
     return tasks
 
 
 def run(
+        loop: Optional[asyncio.AbstractEventLoop] = None,
         lifecycle: Optional[Callable] = None,
         registry: Optional[registries.BaseRegistry] = None,
         standalone: bool = False,
         priority: int = 0,
-        loop: Optional[asyncio.BaseEventLoop] = None,
         peering_name: str = peering.PEERING_DEFAULT_NAME,
         namespace: Optional[str] = None,
 ):
@@ -223,6 +227,7 @@ def run(
     """
     loop = loop if loop is not None else asyncio.get_event_loop()
     tasks = create_tasks(
+        loop=loop,
         lifecycle=lifecycle,
         registry=registry,
         standalone=standalone,

kopf/reactor/registries.py

@@ -12,17 +12,25 @@
 of the handlers to be executed on each reaction cycle.
 """
 import abc
-import collections
 import functools
 from types import FunctionType, MethodType
-from typing import MutableMapping
+from typing import MutableMapping, NamedTuple, Text, Optional, Tuple, Callable
 
 
 # An immutable reference to a custom resource definition.
-Resource = collections.namedtuple('Resource', 'group version plural')
+class Resource(NamedTuple):
+    group: Text
+    version: Text
+    plural: Text
 
 # A registered handler (function + event meta info).
-Handler = collections.namedtuple('Handler', 'fn id event field timeout initial')
+class Handler(NamedTuple):
+    fn: Callable
+    id: Text
+    event: Text
+    field: Optional[Tuple[str, ...]]
+    timeout: Optional[float] = None
+    initial: Optional[bool] = None
 
 
 class BaseRegistry(metaclass=abc.ABCMeta):

kopf/utilities/__init__.py

@@ -0,0 +1,5 @@
+"""
+General-purpose helpers not related to the framework itself
+(neither to the reactor nor to the engines nor to the structs),
+which are used to prepare and control the runtime environment.
+"""

tests/cli/conftest.py

@@ -82,12 +82,12 @@ def invoke(runner):
 
 @pytest.fixture()
 def login(mocker):
-    return mocker.patch('kopf.config.login')
+    return mocker.patch('kopf.clients.auth.login')
 
 
 @pytest.fixture()
 def preload(mocker):
-    return mocker.patch('kopf.reactor.loading.preload')
+    return mocker.patch('kopf.utilities.loaders.preload')
 
 
 @pytest.fixture()

tests/cli/test_help.py

@@ -1,7 +1,7 @@
 
 
 def test_help_in_root(invoke, mocker):
-    login = mocker.patch('kopf.config.login')
+    login = mocker.patch('kopf.clients.auth.login')
 
     result = invoke(['--help'])
 
@@ -15,8 +15,8 @@ def test_help_in_root(invoke, mocker):
 
 
 def test_help_in_subcommand(invoke, mocker):
-    login = mocker.patch('kopf.config.login')
-    preload = mocker.patch('kopf.reactor.loading.preload')
+    login = mocker.patch('kopf.clients.auth.login')
+    preload = mocker.patch('kopf.utilities.loaders.preload')
     real_run = mocker.patch('kopf.reactor.queueing.run')
 
     result = invoke(['run', '--help'])

tests/cli/test_login.py

@@ -1,7 +1,7 @@
 import kubernetes
 import pytest
 
-from kopf.config import login, LoginError
+from kopf.clients.auth import login, LoginError
 
 
 @pytest.fixture(autouse=True)