Upgrade kubernetes_service_patch library (canonical#23)

lib/charms/observability_libs/v1/kubernetes_service_patch.py

@@ -89,6 +89,26 @@ def __init__(self, *args):
     # ...
 ```
 
+Bound with custom events by providing `refresh_event` argument:
+For example, you would like to have a configurable port in your charm and want to apply
+service patch every time charm config is changed.
+
+```python
+from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch
+from lightkube.models.core_v1 import ServicePort
+
+class SomeCharm(CharmBase):
+  def __init__(self, *args):
+    # ...
+    port = ServicePort(int(self.config["charm-config-port"]), name=f"{self.app.name}")
+    self.service_patcher = KubernetesServicePatch(
+        self,
+        [port],
+        refresh_event=self.on.config_changed
+    )
+    # ...
+```
+
 Additionally, you may wish to use mocks in your charm's unit testing to ensure that the library
 does not try to make any API calls, or open any files during testing that are unlikely to be
 present, and could break your tests. The easiest way to do this is during your test `setUp`:
@@ -105,7 +125,7 @@ def setUp(self, *unused):
 
 import logging
 from types import MethodType
-from typing import List, Literal
+from typing import List, Literal, Optional, Union
 
 from lightkube import ApiError, Client
 from lightkube.core import exceptions
@@ -114,7 +134,7 @@ def setUp(self, *unused):
 from lightkube.resources.core_v1 import Service
 from lightkube.types import PatchType
 from ops.charm import CharmBase
-from ops.framework import Object
+from ops.framework import BoundEvent, Object
 
 logger = logging.getLogger(__name__)
 
@@ -126,7 +146,7 @@ def setUp(self, *unused):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 3
+LIBPATCH = 7
 
 ServiceType = Literal["ClusterIP", "LoadBalancer"]
 
@@ -138,11 +158,13 @@ def __init__(
         self,
         charm: CharmBase,
         ports: List[ServicePort],
-        service_name: str = None,
+        service_name: Optional[str] = None,
         service_type: ServiceType = "ClusterIP",
-        additional_labels: dict = None,
-        additional_selectors: dict = None,
-        additional_annotations: dict = None,
+        additional_labels: Optional[dict] = None,
+        additional_selectors: Optional[dict] = None,
+        additional_annotations: Optional[dict] = None,
+        *,
+        refresh_event: Optional[Union[BoundEvent, List[BoundEvent]]] = None,
     ):
         """Constructor for KubernetesServicePatch.
 
@@ -158,6 +180,9 @@ def __init__(
             additional_selectors: Selectors to be added to the kubernetes service (by default only
                 "app.kubernetes.io/name" is set to the service name)
             additional_annotations: Annotations to be added to the kubernetes service.
+            refresh_event: an optional bound event or list of bound events which
+                will be observed to re-apply the patch (e.g. on port change).
+                The `install` and `upgrade-charm` events would be observed regardless.
         """
         super().__init__(charm, "kubernetes-service-patch")
         self.charm = charm
@@ -176,15 +201,24 @@ def __init__(
         # Ensure this patch is applied during the 'install' and 'upgrade-charm' events
         self.framework.observe(charm.on.install, self._patch)
         self.framework.observe(charm.on.upgrade_charm, self._patch)
+        self.framework.observe(charm.on.update_status, self._patch)
+
+        # apply user defined events
+        if refresh_event:
+            if not isinstance(refresh_event, list):
+                refresh_event = [refresh_event]
+
+            for evt in refresh_event:
+                self.framework.observe(evt, self._patch)
 
     def _service_object(
         self,
         ports: List[ServicePort],
-        service_name: str = None,
+        service_name: Optional[str] = None,
         service_type: ServiceType = "ClusterIP",
-        additional_labels: dict = None,
-        additional_selectors: dict = None,
-        additional_annotations: dict = None,
+        additional_labels: Optional[dict] = None,
+        additional_selectors: Optional[dict] = None,
+        additional_annotations: Optional[dict] = None,
     ) -> Service:
         """Creates a valid Service representation.
 
@@ -276,9 +310,8 @@ def _is_patched(self, client: Client) -> bool:
         except ApiError as e:
             if e.status.code == 404 and self.service_name != self._app:
                 return False
-            else:
-                logger.error("Kubernetes service get failed: %s", str(e))
-                raise
+            logger.error("Kubernetes service get failed: %s", str(e))
+            raise
 
         # Construct a list of expected ports, should the patch be applied
         expected_ports = [(p.port, p.targetPort) for p in self.service.spec.ports]