Django中为ApiPartner模型绑定特定函数实现通用get_posts方法

针对你的需求，这里提供几种实用的实现方式，让你可以通过ApiPartner实例调用统一的get_posts()方法，自动匹配对应合作方的数据获取逻辑：

方案一：新增类型字段+模型方法分发 #

先给ApiPartner模型添加一个标记合作方类型的字段，然后在模型的get_posts方法里根据类型调用对应的函数。

修改模型 #

from django.db import models
from django.utils.translation import gettext_lazy as _

class ApiPartner(models.Model):
    class PartnerType(models.TextChoices):
        YOUTUBE = "YOUTUBE", _("Youtube")
        INSTAGRAM = "INSTAGRAM", _("Instagram")
        # 后续新增合作方直接在这里加选项

    id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=30, verbose_name=_('Name'))
    partner_type = models.CharField(
        max_length=20,
        choices=PartnerType.choices,
        verbose_name=_('Partner Type'),
        null=False,
        blank=False
    )
    api_key = models.CharField(max_length=50, verbose_name=_('API key'), null=True)
    secret_key = models.CharField(max_length=50, verbose_name=_('Secret key'), null=True)
    client_key = models.CharField(max_length=50, verbose_name=_('Client key'), null=True)
    endpoint = models.CharField(max_length=50, verbose_name=_('Endpoint'), null=True)
    logo = models.ImageField(upload_to='logos/', null=True)

    def get_posts(self):
        # 根据合作方类型调用对应函数
        if self.partner_type == self.PartnerType.YOUTUBE:
            if not self.endpoint or not self.api_key:
                raise ValueError("Youtube合作方需要填写endpoint和api_key")
            return get_youtube_posts(self.endpoint, self.api_key)
        elif self.partner_type == self.PartnerType.INSTAGRAM:
            if not self.endpoint or not self.api_key or not self.secret_key:
                raise ValueError("Instagram合作方需要填写endpoint、api_key和secret_key")
            return get_instagram_posts(self.endpoint, self.api_key, self.secret_key)
        else:
            raise NotImplementedError(f"未实现{self.partner_type}类型的获取方法")

使用方式 #

# 获取某个Youtube合作方实例
youtube_partner = ApiPartner.objects.get(partner_type=ApiPartner.PartnerType.YOUTUBE)
# 调用统一方法获取数据
posts = youtube_partner.get_posts()

方案二：函数映射字典（更简洁的分发逻辑） #

如果后续要新增更多合作方，方案一的if-elif会越来越臃肿，用字典映射类型和对应函数的方式更易维护：

重构函数+映射字典 #

# 先重构获取函数，统一接收实例参数
def get_youtube_posts(partner):
    if not partner.endpoint or not partner.api_key:
        raise ValueError("Youtube合作方需要填写endpoint和api_key")
    results = list(requests.get(f"{partner.endpoint}?key={partner.api_key}").json())
    return results[:50]

def get_instagram_posts(partner):
    if not partner.endpoint or not partner.api_key or not partner.secret_key:
        raise ValueError("Instagram合作方需要填写endpoint、api_key和secret_key")
    return requests.get(f"{partner.endpoint}?key={partner.api_key}&secret={partner.secret_key}")

# 定义全局的函数映射
POST_FETCHERS = {
    ApiPartner.PartnerType.YOUTUBE: get_youtube_posts,
    ApiPartner.PartnerType.INSTAGRAM: get_instagram_posts,
    # 新增合作方时直接在这里添加键值对
}

# 模型的get_posts方法简化
class ApiPartner(models.Model):
    # ... 其他字段和PartnerType类不变 ...

    def get_posts(self):
        fetcher = POST_FETCHERS.get(self.partner_type)
        if not fetcher:
            raise NotImplementedError(f"未实现{self.partner_type}类型的获取方法")
        return fetcher(self)

方案三：代理模型（适合固定合作方类型的场景） #

如果每个合作方类型有独立的业务逻辑，还可以用Django的代理模型，每个类型对应一个代理模型，重写get_posts方法：

定义代理模型 #

# 基础模型不变，去掉partner_type字段
class ApiPartner(models.Model):
    id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=30, verbose_name=_('Name'))
    api_key = models.CharField(max_length=50, verbose_name=_('API key'), null=True)
    secret_key = models.CharField(max_length=50, verbose_name=_('Secret key'), null=True)
    client_key = models.CharField(max_length=50, verbose_name=_('Client key'), null=True)
    endpoint = models.CharField(max_length=50, verbose_name=_('Endpoint'), null=True)
    logo = models.ImageField(upload_to='logos/', null=True)

# Youtube代理模型
class YoutubePartner(ApiPartner):
    class Meta:
        proxy = True

    def get_posts(self):
        if not self.endpoint or not self.api_key:
            raise ValueError("Youtube合作方需要填写endpoint和api_key")
        return get_youtube_posts(self.endpoint, self.api_key)

# Instagram代理模型
class InstagramPartner(ApiPartner):
    class Meta:
        proxy = True

    def get_posts(self):
        if not self.endpoint or not self.api_key or not self.secret_key:
            raise ValueError("Instagram合作方需要填写endpoint、api_key和secret_key")
        return get_instagram_posts(self.endpoint, self.api_key, self.secret_key)

使用方式 #

# 查询Youtube合作方
youtube_partner = YoutubePartner.objects.get(name="Youtube")
posts = youtube_partner.get_posts()

# 查询Instagram合作方
instagram_partner = InstagramPartner.objects.get(name="Instagram")
posts = instagram_partner.get_posts()

内容的提问来源于stack exchange，提问作者Michele Candura