e_lib.product_log

  1# -*- coding: utf-8 -*-
  2"""e_lib.product_log
  3
  4Python client for the Products REST API (application usage statistics).
  5
  6Mirrors the Java ``ru.gentoos.ProductsLog`` library.  Reads
  7``PRODUCTS_HOST``, ``PRODUCTS_USER``, ``PRODUCTS_PASS`` from the
  8environment (or explicit constructor arguments) and provides helpers
  9to register a product, post status messages and log application start /
 10shutdown events.
 11
 12If any required credential is missing the client disables itself
 13silently (logs a warning, all public methods become no-ops) so that
 14development environments without the service never crash the application
 15boot.
 16
 17Example::
 18
 19    from e_lib.product_log import ProductsLog
 20
 21    log = ProductsLog("MyApp", "1.0.0")
 22    log.post_started()
 23    ...
 24    log.post_shutdown()
 25"""
 26
 27from __future__ import annotations
 28
 29import logging
 30import os
 31from typing import Any, Dict, Optional
 32from urllib.parse import quote
 33
 34import requests
 35
 36
 37class ProductsLog:
 38    """Client for the Products REST API.
 39
 40    Parameters
 41    ----------
 42    product_name:
 43        Logical name of the product (must be unique in Products).
 44    product_version:
 45        Version string attached to every stat record.
 46    base_url:
 47        Base URL of the Products service (``PRODUCTS_HOST`` env fallback).
 48    username:
 49        Basic-auth username (``PRODUCTS_USER`` env fallback).
 50    password:
 51        Basic-auth password (``PRODUCTS_PASS`` env fallback).
 52    timeout:
 53        HTTP timeout in seconds for every request.
 54    logger:
 55        Logger for messages.  Projects may inject their own (e.g. a
 56        loguru logger) so that all output goes through a single sink.
 57    """
 58
 59    def __init__(
 60        self,
 61        product_name: str,
 62        product_version: str = "UNKNOWN",
 63        base_url: Optional[str] = None,
 64        username: Optional[str] = None,
 65        password: Optional[str] = None,
 66        timeout: float = 5.0,
 67        logger: Optional[Any] = None,
 68    ) -> None:
 69        self.product_name = product_name
 70        self.product_version = product_version
 71        self.timeout = timeout
 72        self._logger = logger or logging.getLogger(__name__)
 73
 74        self.base_url = (base_url or os.getenv("PRODUCTS_HOST", "")).rstrip("/")
 75        self.username = username or os.getenv("PRODUCTS_USER", "")
 76        self.password = password or os.getenv("PRODUCTS_PASS", "")
 77
 78        self._enabled = bool(self.base_url and self.username and self.password)
 79        if not self._enabled:
 80            self._logger.warning(
 81                "ProductsLog: PRODUCTS_HOST / PRODUCTS_USER / PRODUCTS_PASS "
 82                "are not all set — statistics reporting is disabled."
 83            )
 84            return
 85
 86        self._auth = (self.username, self.password)
 87        self._product_id: Optional[int] = None
 88        self._product_found = False
 89
 90    # ------------------------------------------------------------------
 91    # Internal helpers
 92    # ------------------------------------------------------------------
 93
 94    def _ensure_product(self) -> Optional[int]:
 95        """Return the product id, fetching or registering as needed."""
 96        if self._product_id is not None:
 97            return self._product_id
 98
 99        self._product_id = self._find_product_id()
100        if self._product_id is None:
101            self._product_id = self._register_product_id()
102        return self._product_id
103
104    def _find_product_id(self) -> Optional[int]:
105        """GET /products/byName/{name}.  Returns the id or None."""
106        url = f"{self.base_url}/products/byName/{quote(self.product_name, safe='')}"
107        try:
108            resp = requests.get(url, auth=self._auth, timeout=self.timeout)
109            if resp.status_code == 200:
110                return resp.json().get("id")
111            return None
112        except requests.RequestException as exc:
113            self._logger.warning(
114                "ProductsLog: failed to look up product %r: %s",
115                self.product_name,
116                exc,
117            )
118            return None
119
120    def _register_product_id(self) -> Optional[int]:
121        """POST /products — register a new product.  Returns the new id."""
122        url = f"{self.base_url}/products"
123        try:
124            resp = requests.post(
125                url,
126                json={"name": self.product_name},
127                auth=self._auth,
128                timeout=self.timeout,
129            )
130            if resp.status_code in (200, 201):
131                return resp.json().get("id")
132            self._logger.warning(
133                "ProductsLog: failed to register product %r (HTTP %d)",
134                self.product_name,
135                resp.status_code,
136            )
137            return None
138        except requests.RequestException as exc:
139            self._logger.warning(
140                "ProductsLog: failed to register product %r: %s",
141                self.product_name,
142                exc,
143            )
144            return None
145
146    # ------------------------------------------------------------------
147    # Public API
148    # ------------------------------------------------------------------
149
150    def post_message(self, comment: str) -> Optional[Dict[str, Any]]:
151        """Post a stat record to the Products API.
152
153        Parameters
154        ----------
155        comment:
156            Free-text message (e.g. ``"Program started"``).
157
158        Returns
159        -------
160        The parsed JSON response body (``{"id": ..., "product": ...,
161        "version": ..., "comment": ..., "created": ...}``) or ``None``
162        on failure.
163        """
164        if not self._enabled:
165            return None
166
167        product_id = self._ensure_product()
168        if product_id is None:
169            return None
170
171        payload = {
172            "product": {"id": product_id},
173            "version": self.product_version,
174            "comment": comment,
175        }
176        url = f"{self.base_url}/stats"
177        try:
178            resp = requests.post(
179                url,
180                json=payload,
181                auth=self._auth,
182                timeout=self.timeout,
183            )
184            if resp.status_code in (200, 201):
185                return resp.json()
186            self._logger.warning(
187                "ProductsLog: post_message failed (HTTP %d): %s",
188                resp.status_code,
189                resp.text,
190            )
191            return None
192        except requests.RequestException as exc:
193            self._logger.warning("ProductsLog: post_message failed: %s", exc)
194            return None
195
196    def post_started(self) -> Optional[Dict[str, Any]]:
197        """Convenience — posts ``"Program started"``."""
198        return self.post_message("Program started")
199
200    def post_shutdown(self) -> Optional[Dict[str, Any]]:
201        """Convenience — posts ``"Program shutdown"``."""
202        return self.post_message("Program shutdown")