python - 如何为具有相同结构的方法创建泛型方法？

Question

我在 CommentsService 类中有以下方法:

async def background_job_auto_approve(self):
    while True:
        new = get_comments_by_status(CommentStatus.NEW.value)
        pending = get_comments_by_status(CommentStatus.PENDING.value)
        all = new + pending
        for comment in all:
            if check_it_auto_approve(item=comment):
                await self.auto_approve(comment_id=comment['comment_id'],
                                        alert_id=comment['alert_id'])
                yield comment
        await asyncio.sleep(self.check_expire_seconds)

但我的AlertsService中有完全相同的方法:

async def background_job_auto_approve(self):
    while True:
        new = get_alerts_by_status(AlertStatus.NEW.value)
        pending = get_alerts_by_status(AlertStatus.PENDING.value)
        all = new + pending
        for alert in all:
            if check_it_auto_approve(item=alert):
                await self.auto_approve(alert_id=alert['alert_id'])
                yield alert
        await asyncio.sleep(self.check_expire_seconds)

如何避免代码重复？我对这些类中的其他方法也有同样的问题。

Answer 1

如果没有其目的的其余代码库/上下文，很难想出一种真正通用的方法来执行此操作。但是，如果您可以巩固您似乎正在使用的模式(即 [objectname]Status、[objectname]_id、get_[objectname]s_by_status)，这将工作正常。我建议为方法添加一个额外的类，例如 auto_approve

class AutoApprovalLoop(object):
    def __init__(self, method_ptr=None, cls_ptr=None, approval_key=None):
        self.method_ptr = method_ptr     # one of get_comments_by_status or get_alerts_by_status
        self.cls_ptr = cls_ptr           # one of AlertStatus or CommentStatus
        self.approval_key = approval_key # one of 'comment_id' or 'alert_id'

    def loop(self):
        while True:
            new = self.method_ptr(self.cls_ptr.NEW.value)
            pending = self.method_ptr(self.cls_ptr.PENDING.value) 
            all = new + pending
            for item in all:
                if check_it_auto_approve(item=item)
                await self.auto_approve(**{self.approval_key: item[self.approval_key]})
                yield item
            await asyncio.sleep(self.check_expire_seconds)