Egor fureunoir Gorbunov
9898212855
Fixes: 1) Fixed missing `exc_info` flag in critical and error logs to provide richer error context; 2) Addressed redundant code and unused imports in the logging logic for cleaner execution. Extra: Refactored `LOGGING` configuration for improved readability and runtime adaptability; optimized related log-handling logic throughout the codebase.
39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
import logging
|
|
from contextlib import suppress
|
|
from typing import Type
|
|
|
|
from celery import current_task
|
|
from celery.utils.log import get_task_logger
|
|
|
|
from core.models import Vendor
|
|
from core.vendors import AbstractVendor
|
|
from evibes.utils.misc import create_object
|
|
|
|
sync_logger = logging.getLogger(__name__)
|
|
async_logger = get_task_logger(__name__)
|
|
|
|
|
|
def _in_celery_task() -> bool:
|
|
task = current_task
|
|
with suppress(Exception):
|
|
return bool(task and getattr(task, "request", None) and getattr(task.request, "id", None))
|
|
return False
|
|
|
|
|
|
def get_vendors_integrations(name: str | None = None) -> list[Type[AbstractVendor]]:
|
|
vendors_integrations: list[Type[AbstractVendor]] = []
|
|
|
|
vendors = Vendor.objects.filter(is_active=True, integration_path__isnull=False)
|
|
if name:
|
|
vendors = vendors.filter(name=name)
|
|
|
|
logger = async_logger if _in_celery_task() else sync_logger
|
|
|
|
for vendor in vendors:
|
|
try:
|
|
module_name, class_name = vendor.integration_path.rsplit(".", 1)
|
|
vendors_integrations.append(create_object(module_name, class_name))
|
|
except Exception as e:
|
|
logger.warning("Couldn't load integration for vendor %s: %s", vendor.name, e)
|
|
|
|
return vendors_integrations
|