当前位置：首页 > news >正文

工厂模式与抽象工厂

news 2025/8/29 4:05:22

原理：逻辑和业务全部封装不需要细节只要结果

示例：

# 简单工厂
class SimpleFactory:# 产品@staticmethoddef product(name):return nameif __name__ == "__main__":product = SimpleFactory.product("Gitee")print(product)

装饰器@staticmethod，普通方法，类对象实例对象都可应用
装饰的方法类比产品，工厂生产产品，只要最后结果

抽象工厂实例：

from abc import ABCMeta, abstractmethod  # 抽象工厂class TestMeta(metaclass=ABCMeta):  继承metaclass=ABCMetadef __init__(self):self.a = 1@abstractmethoddef fun(self):passclass TestOne(TestMeta):def __init__(self):super().__init__()  # 调用父类的init# TestMeta.__init__()self.b = 1# 重写抽象方法def fun(self):print("拼接三方登录的url")

抽象工厂类似于生活中的橘子，橘子的种类繁多，但橘子本体的一些口感(方法)是其他种类必须所拥有的，也就是其他类继承抽象工厂类，必须重写已有的方法，否则报错
pass是占位符，抽象工厂中的方法要用装饰器@abstractmethod定义，只定义框架，不写具体逻辑

抽象工厂完成第三方登录基本方法：

class LoginProvide(metaclass=ABCMeta):# 跳转拼接@abstractmethoddef get_url(self):pass# 获取token@abstractmethodasync def get_token(self, code):pass# 获取用户信息@abstractmethodasync def get_info(self, token):pass# 用户信息存储@abstractmethodasync def set_user(self, user):pass

gitee登录实例

class GiteeProvider(LoginProvide):def __repr__(self):return "Gitee登录"def get_url(self):url = SITE_TYPE["Gitee"]["url"]  # 跳到授权地址return urlasync def get_token(self, code):# 通过授权码请求token# 异步发送网络请求，防止阻塞client_id = SITE_TYPE["Gitee"]["client_id"]client_secret = SITE_TYPE["Gitee"]["client_secret"]redirect_uri = SITE_TYPE["Gitee"]["redirect_uri"]async with httpx.AsyncClient() as client:res = await client.post(url=f"https://gitee.com/oauth/token?grant_type=authorization_code&code={code}&client_id={client_id}&redirect_uri={redirect_uri}&client_secret={client_secret}",timeout=20)res = res.json()# 获取gitee的access_token refresh_tokenreturn res["access_token"]async def get_info(self, token):async with httpx.AsyncClient() as client:res = await client.get(SITE_TYPE["Gitee"]["gitee_get_info_uri"] + token, timeout=20)res = res.json()user = {"gitee_id": res["id"],"gitee_username": res["name"]}return userasync def set_user(self, user):gitee_id = user["gitee_id"]gitee_username = user["gitee_username"]username = (str(gitee_id) + "_" + str(gitee_username))try:# 曾经使用过gitee账号await db.get(UserModel.select().where((UserModel.username == username)& (UserModel.site_type == SITE_TYPE["Gitee"]["num"])))except Exception as e:print(e.args)await db.create(UserModel,username=username,site_type=SITE_TYPE["Gitee"]["num"])finally:mj = MyJwt()# 生成带有生命周期的tokentoken = mj.encode_date({"username": username,"site_type": SITE_TYPE["Gitee"]["num"]})# 生成refresh_token  redis里面存refresh_token = await mj.set_refresh_token(f"{gitee_id}_{gitee_username}")# 中间缓存页url = SITE_TYPE["Gitee"]["gitee_center_uri"] + f"?gitee_refresh_token={refresh_token}&gitee_token={token}&username={username}"return url

Gitee简化版：

# gitee登录简化版
class GiteeFactory(BaseHandler):def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.gitee = GiteeProvider()async def get(self):code = self.get_argument("code")access_token = await self.gitee.get_token(code)user = await self.gitee.get_info(access_token)url = await self.gitee.set_user(user)return self.redirect(url)# 路由
urlpatterns = [(r"/gitee_back/", GiteeFactory),  # gitee
]

总结：