Egor fureunoir Gorbunov
202964aea6
Fixes: None; Extra: 1) Refactor vendor filtering logic for clarity; 2) Add imports for logging, contextlib, and Celery utilities;
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("django")
|
|
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
|