使用Redis实现简单的事件驱动架构 「DDD、事件溯源和一致性哈希」

from typing import List
from .services.base import ServiceBase
from sqlalchemy.session import Session

class PostService(ServiceBase)
    def __init__(self, thread_id: UUID, params:   PostCreateAPIParams, db_session: Session):
        self.thread_id = thread_id
        self.params = params
        self.user: Union[User, None] = None
        self.post: Union[Post, None] =  None
        self.db_session = db_session
        self.errors: List[str] = []
        self.error_code: Union[str, None] = None
 
     async def __call__(self)
        return awAIt self.invoke()

     async def invoke(self):
         await self.find_thread()
         await self.verify_author()
         await self.create_post()
         await self.build_response_dao()
         await self.trigger_events()
         return self   

     async def find_thread(self):
    # truncated for brevity
    pass
    
    
     async def trigger_events(self):
    user_dao: UserDAO = UserDAO.from_orm(self.user) if self.user else None
        post_dao: PostDAO = PostDAO.from_orm(self.post) if self.post else None
        thread_dao: ThreadDAO = ThreadDAO.from_orm(self.thread) if self.thread else None

        if await self.has_errors:
            event_dao = PostCreatedEventDAO(
                user=user_dao,
                thread=thread_dao,
                post=post_dao,
                params=self.params,
                errors=self.errors,
                error_code= self.error_code
            )
        else:
            event_dao = PostCreationFailedEventDAO(
        user=user_dao,
                thread=thread_dao,
                params=self.params,
                post=post_dao
                errors=self.errors,
                error_code= self.error_code
            )     
                 
       partition_key = (
       str(self.thread.id) if self.thread else "PostCreationFailedEvent"
        )
       await SystemEventService.trigger(partition_key=partition_key, event_dao=event_dao, db_session=self.db_session)
       return self

{
   "event_name": "PostCreatedEvent", 
   "event_id": "0fb6a4d4-ae65-4f18-be44-edb9ace6b5bb",
   "event_version": "v1.0",
   "time": "2022-09-03T04:16:59.294509+00:00",
   "payload": {
           "user": { "user_id":"1a1269ee-6b6f-4325-8562-cb169a68e7b3", "is_blocked": false, "first_name": "Siddharth", "last_name": "R", "email": "sid@........},
           "post" :{ "post_id": "fa3e7b12-4908-4d53-be11-629e6f47ae90", "thread_id": "666d0404-0756-4b93-892f-19d9b4b25a99", ...... },
           "thread": { "thread_id": "666d0404-0756-4b93-892f-19d9b4b25a99", .....},
           "params": { .... },
           "errors": [ ... ],
           "error_code": ""
    },
    "created_at": "2022-09-03T04:16:59.294"
    "logged_at": "2022-09-03T04:16:59.294"
}

@staticmethod
async def trigger(
    partition_key: str, event_dao: SystemEventDAO
):    

    try:
        worker: SystemEventPartitionConfig = await SystemEventsService._get_worker(
            partition_key=partition_key
        )
        worker_func = getattr(system_events_workers, worker.worker_name)
        log_info(msg=f"Trigger called with worker: {worker.worker_name}")
        worker_func.delay(event_dao=event_dao.json())
    except (OperationalError, ConnectionError) as e:
        log_error(msg=f"[redisError] {e}", e=e, method="trigger", loc=f"{__name__}")
    except Exception as e:
        log_error(
            msg=f"SystemEventError: {e}", e=e, method="trigger", loc=f"{__name__}"
        )
    return

@staticmethod
async def _get_worker(partition_key: str) -> SystemEventPartitionConfig:
    """For a given string it returns the worker that should process the event by running it through a murmurr hashing
    function and uses that to fetch the nodes from the continuum"""

    node = ring.get(key=partition_key)
    nodename = node.get("nodename", None)

    if not nodename:
        raise ValueError("Could not find a node in the continuum for key {node}")

    node_config = continuum.get(nodename, None)
    if not node_config:
        raise ValueError(
            "Could not find a node in the continuum for key {nodename}"
        )

    config_attrs = {"partition_key": partition_key, "partition_id": nodename}
    config_attrs = {**node_config, **config_attrs}
    return SystemEventPartitionConfig(**config_attrs)

async def create_system_event(
    task_type, event_dao: SystemEventDAO, db_session: Session = None
):
    if not db_session:
        db_session = get_session()

    system_event: Union[SystemEvent, None] = None

    try:
        if event_dao.event_name not in SYSTEM_GENERATED_REQUEST_EVENTS:
            system_event = await SystemEventsService.create(
                event_dao=event_dao, db_session=db_session
            )

            if system_event:
                log_info(
                    msg=f"system_event with id: {system_event.id} created for event_name: {system_event.event_name}"
                )
                event_dao.id = system_event.id
        else:
            log_info(
                msg=f"system_generated_request_event with name: {event_dao.event_name} ready for processing."
            )
        await EventHandler.process(event_dao=event_dao, db_session=db_session)
    except Exception as e:
        log_error(
            msg=f"Error handling events: {event_dao.event_name}: {e} \n {traceback.print_exc()}"
        )
        capture_exc(error=e)
    finally:
        db_session.close()
    return system_event

handlers = [
    PostInsightsGeneratorEventHandler,
    ThreadActivityManagerEventHandler,
    SpamDetectionEventHandler,
    #...,
    #....,
    ImageResizerEventHandler,
]

class EventHandler:
    @staticmethod
    async def process(event_dao: SystemEventDAO, db_session: Session = None):
        log_info(msg=f'Event: {event_dao.event_name} arrived')
        
        for event_handler in handlers:
            await event_handler.process(event_dao=event_dao, db_session=db_session)
        return

class PostInsightsGeneratorEventHandler(event_dao: SystemEventDAO, db_session: Session):
    if not event_dao.event_name == "PostCreatedEvent" :
         return 

    log_info(msg=f"PostCreatedEventHandler called with {event_dao.event_id}")
    # Do whatever you need to here