完成三方登录三步骤:定义抽象类,定义实现具体功能类,定义视图类
流程图:
定义抽象类:
首先导入模块:
# 导入
from abc import ABCMeta,abstractmethod
再定义抽象类:
# 抽象类 三方登录抽象类
class IdProvider(metaclass=ABCMeta):# 生成跳转的url
@abstractmethoddef get_url(self):pass# 获取token
@abstractmethoddef get_token(self, code):pass# 获取用户信息
@abstractmethoddef get_user(self, token, uid):pass# 保存用户信息
@abstractmethoddef save_user(self,user):pass
定义实现具体功能类:
1.获取token的路由:
请求用户授权Token
2.获取授权过的token:
获取授权过的Access Token
3.获取授权过的用户信息
获取用户信息
# 实现具体功能类
class SinaProvider(IdProvider):def __init__(self):self.client_id = "3440988304"self.client_secret = "15dde1e8af983da29490e5ac5de10f55"self.call_back = "http://127.0.0.1:8000/weibo_back/"def get_url(self):return "https://api.weibo.com/oauth2/authorize?client_id=%s&redirect_uri=%s" % (self.client_id,self.call_back)def get_token(self, code):data = {"client_id":self.client_id,"client_secret":self.client_secret,"grant_type":"authorization_code","code":code,"redirect_uri":self.call_back}res = requests.post('https://api.weibo.com/oauth2/access_token', data=data)print(res.text)return res.json()["access_token"],res.json()["uid"]def get_user(self, token, uid):data = {"access_token": token, "uid": uid}res = requests.get("https://api.weibo.com/2/users/show.json", params=data, timeout=25)print(res.text)return {"name": res.json()["name"], "avtar": res.json()["profile_image_url"]}
微博API接口文档: 获取用户信息
定义视图类:
# 视图类
class Sina(BaseHandler):def __init__(self,*args,**kwargs):super(Sina,self).__init__(*args,**kwargs)# 实例化self.sina = SinaProvider()async def post(self):# 获取urlurl = self.sina.get_url()self.finish({"errcode":0,"url":url})async def get(self):code = self.get_argument('code',None)# 获取新浪tokentoken,uid = self.sina.get_token(code)# 获取用户信息user = self.sina.get_user(token,uid)self.finish({"user":user})
打印出的结果为:
然后将获得的用户信息存入数据库:
导入database:
from db import database
写在具体功能类下:
# 实现具体功能类
class SinaProvider(IdProvider):def __init__(self):self.client_id = "3440988304"self.client_secret = "15dde1e8af983da29490e5ac5de10f55"self.call_back = "http://127.0.0.1:8000/weibo_back/"self.database = peewee_async.Manager(database)def get_url(self):return "https://api.weibo.com/oauth2/authorize?client_id=%s&redirect_uri=%s" % (self.client_id,self.call_back)def get_token(self, code):data = {"client_id":self.client_id,"client_secret":self.client_secret,"grant_type":"authorization_code","code":code,"redirect_uri":self.call_back}res = requests.post('https://api.weibo.com/oauth2/access_token', data=data)print(res.text)return res.json()["access_token"],res.json()["uid"]def get_user(self, token, uid):data = {"access_token": token, "uid": uid}res = requests.get("https://api.weibo.com/2/users/show.json", params=data, timeout=25)print(res.text)return {"name": res.json()["name"], "avtar": res.json()["profile_image_url"]}async def save_user(self,user):try:user = await self.database.get(UserModel.select().where((UserModel.email == user["name"]) & (UserModel.cate == 2)))except Exception as e:user = await self.database.create(UserModel, email=user["name"], password="123", state=1,avtar=user["avtar"], cate=2)return user
保存用户信息,并且重定向:
# 视图类
class Sina(BaseHandler):def __init__(self,*args,**kwargs):super(Sina,self).__init__(*args,**kwargs)# 实例化self.sina = SinaProvider()async def post(self):# 获取urlurl = self.sina.get_url()self.finish({"errcode":0,"url":url})async def get(self):code = self.get_argument('code',None)# 获取新浪tokentoken,uid = self.sina.get_token(code)# 获取用户信息user = self.sina.get_user(token,uid)# 保存用户user = await self.sina.save_user(user)# 重定向# 生成jwtmyjwt = MyJwt()# 重定向self.redirect("http://localhost:8080/tranfer?email=" + user.email + "&token=" + myjwt.encode({"id": user.id}))