import abc import typing from celery import shared_task from atlas_um import extensions from atlas_um.consts import SystemEvents class BaseAnalyticsProcessor: """ Interface for analytics processors. Processors are observers that intended to handle analytics events in any needed way, e.g. send to services like Amplitude, etc. """ @abc.abstractmethod def process(self, event: SystemEvents, context: typing.Mapping, **kwargs): pass class AsyncAmplitudeProcessor(BaseAnalyticsProcessor): """ Processor for sending events to Amplitude in the async way with Celery task. """ def process(self, event: SystemEvents, context: typing.Mapping, **kwargs): event_data = { "event_type": event.value, "user_id": kwargs.get("user_id") or context.get("user_id"), "event_properties": {**kwargs, **context}, } self.send_to_amplitude.delay(event_data) @staticmethod @shared_task def send_to_amplitude(event_data: typing.Mapping): extensions.amplitude.send_event(event_data)