mirror of
https://github.com/TandoorRecipes/recipes.git
synced 2026-01-01 04:10:06 -05:00
Merge pull request #3215 from mikhail5555/enable-global-logging-and-more-connector-logging
enable logging for whole project and add more logging to connectors
This commit is contained in:
@@ -5,6 +5,7 @@ import threading
|
||||
from asyncio import Task
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from logging import Logger
|
||||
from types import UnionType
|
||||
from typing import List, Any, Dict, Optional, Type
|
||||
|
||||
@@ -39,10 +40,12 @@ class Work:
|
||||
# 4. Work is marked as consumed, and next entry of the queue is consumed.
|
||||
# Each 'Work' is processed in sequential by the worker, so the throughput is about [workers * the slowest connector]
|
||||
class ConnectorManager:
|
||||
_logger: Logger
|
||||
_queue: queue.Queue
|
||||
_listening_to_classes = REGISTERED_CLASSES | ConnectorConfig
|
||||
|
||||
def __init__(self):
|
||||
self._logger = logging.getLogger("recipes.connector")
|
||||
self._queue = queue.Queue(maxsize=settings.EXTERNAL_CONNECTORS_QUEUE_SIZE)
|
||||
self._worker = threading.Thread(target=self.worker, args=(0, self._queue,), daemon=True)
|
||||
self._worker.start()
|
||||
@@ -65,7 +68,7 @@ class ConnectorManager:
|
||||
try:
|
||||
self._queue.put_nowait(Work(instance, action_type))
|
||||
except queue.Full:
|
||||
logging.info(f"queue was full, so skipping {action_type} of type {type(instance)}")
|
||||
self._logger.info(f"queue was full, so skipping {action_type} of type {type(instance)}")
|
||||
return
|
||||
|
||||
def stop(self):
|
||||
@@ -74,10 +77,12 @@ class ConnectorManager:
|
||||
|
||||
@staticmethod
|
||||
def worker(worker_id: int, worker_queue: queue.Queue):
|
||||
logger = logging.getLogger("recipes.connector.worker")
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
logging.info(f"started ConnectionManager worker {worker_id}")
|
||||
logger.info(f"started ConnectionManager worker {worker_id}")
|
||||
|
||||
# When multiple workers are used, please make sure the cache is shared across all threads, otherwise it might lead to un-expected behavior.
|
||||
_connectors_cache: Dict[int, List[Connector]] = dict()
|
||||
@@ -91,6 +96,8 @@ class ConnectorManager:
|
||||
if item is None:
|
||||
break
|
||||
|
||||
logger.debug(f"received {item.instance=} with {item.actionType=}")
|
||||
|
||||
# If a Connector was changed/updated, refresh connector from the database for said space
|
||||
refresh_connector_cache = isinstance(item.instance, ConnectorConfig)
|
||||
|
||||
@@ -111,7 +118,7 @@ class ConnectorManager:
|
||||
try:
|
||||
connector: Optional[Connector] = ConnectorManager.get_connected_for_config(config)
|
||||
except BaseException:
|
||||
logging.exception(f"failed to initialize {config.name}")
|
||||
logger.exception(f"failed to initialize {config.name}")
|
||||
continue
|
||||
|
||||
if connector is not None:
|
||||
@@ -123,10 +130,12 @@ class ConnectorManager:
|
||||
worker_queue.task_done()
|
||||
continue
|
||||
|
||||
logger.debug(f"running {len(connectors)} connectors for {item.instance=} with {item.actionType=}")
|
||||
|
||||
loop.run_until_complete(run_connectors(connectors, space, item.instance, item.actionType))
|
||||
worker_queue.task_done()
|
||||
|
||||
logging.info(f"terminating ConnectionManager worker {worker_id}")
|
||||
logger.info(f"terminating ConnectionManager worker {worker_id}")
|
||||
|
||||
asyncio.set_event_loop(None)
|
||||
loop.close()
|
||||
|
||||
@@ -17,10 +17,11 @@ class HomeAssistant(Connector):
|
||||
if not config.token or not config.url or not config.todo_entity:
|
||||
raise ValueError("config for HomeAssistantConnector in incomplete")
|
||||
|
||||
self._logger = logging.getLogger(f"recipes.connector.homeassistant.{config.name}")
|
||||
|
||||
if config.url[-1] != "/":
|
||||
config.url += "/"
|
||||
self._config = config
|
||||
self._logger = logging.getLogger("connector.HomeAssistant")
|
||||
|
||||
async def homeassistant_api_call(self, method: str, path: str, data: Dict) -> str:
|
||||
headers = {
|
||||
@@ -37,7 +38,7 @@ class HomeAssistant(Connector):
|
||||
|
||||
item, description = _format_shopping_list_entry(shopping_list_entry)
|
||||
|
||||
logging.debug(f"adding {item=} to {self._config.name}")
|
||||
self._logger.debug(f"adding {item=}")
|
||||
|
||||
data = {
|
||||
"entity_id": self._config.todo_entity,
|
||||
@@ -48,7 +49,7 @@ class HomeAssistant(Connector):
|
||||
try:
|
||||
await self.homeassistant_api_call("POST", "services/todo/add_item", data)
|
||||
except ClientError as err:
|
||||
self._logger.warning(f"[HomeAssistant {self._config.name}] Received an exception from the api: {err=}, {type(err)=}")
|
||||
self._logger.warning(f"received an exception from the api: {err=}, {type(err)=}")
|
||||
|
||||
async def on_shopping_list_entry_updated(self, space: Space, shopping_list_entry: ShoppingListEntry) -> None:
|
||||
if not self._config.on_shopping_list_entry_updated_enabled:
|
||||
@@ -66,7 +67,7 @@ class HomeAssistant(Connector):
|
||||
|
||||
item, _ = _format_shopping_list_entry(shopping_list_entry)
|
||||
|
||||
logging.debug(f"removing {item=} from {self._config.name}")
|
||||
self._logger.debug(f"removing {item=}")
|
||||
|
||||
data = {
|
||||
"entity_id": self._config.todo_entity,
|
||||
@@ -77,7 +78,7 @@ class HomeAssistant(Connector):
|
||||
await self.homeassistant_api_call("POST", "services/todo/remove_item", data)
|
||||
except ClientError as err:
|
||||
# This error will always trigger if the item is not present/found
|
||||
self._logger.debug(f"[HomeAssistant {self._config.name}] Received an exception from the api: {err=}, {type(err)=}")
|
||||
self._logger.debug(f"received an exception from the api: {err=}, {type(err)=}")
|
||||
|
||||
async def close(self) -> None:
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user