oauthlib 在 python falcon 中创建自己的 oauthlib2
oauthlib creating own oauthlib2 in python falcon
我的数据库是这样的
from sqlalchemy.ext.declarative import declarative_base
import os
import time
import datetime
import sys
import uuid
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
from sqlalchemy import *
from sqlalchemy_utils.types.choice import ChoiceType
from sqlalchemy_utils.types.url import URLType
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
fullname = Column(String(250), nullable=False)
password = Column(String(250), nullable=False)
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
class Client(Base):
Grand_choice = [('authorization_code', 'Authorization code')]
Response_choice = [('code', 'Authorization code')]
__tablename__ = 'client'
id = Column(Integer, primary_key=True)
client_id = Column(String(250), default=uuid.uuid4() ,unique=True)
user = Column(Integer, ForeignKey('user.id'))
grant_type = Column(ChoiceType(Grand_choice))
response_type = Column(ChoiceType(Response_choice))
scopes = Column(String(260))
default_scopes = Column(String(260))
redirect_uri = Column(URLType)
default_redirect_uri = Column(URLType)
class Bearer_Token(Base):
__tablename__ = 'Bearer_Token'
id = Column(Integer, primary_key=True)
client = Column(Integer, ForeignKey('client.id'))
user = Column(Integer, ForeignKey('user.id'))
scopes = Column(String(250))
access_token = Column(String(100),unique=True)
refresh_token = Column(String(100),unique=True)
expires_at = Column(DateTime, onupdate=datetime.datetime.now)
class Authorization_Code(Base):
__tablename__ = 'Authorization_code'
id = Column(Integer, primary_key=True)
client = Column(Integer, ForeignKey('client.id'))
user = Column(Integer, ForeignKey('user.id'))
scopes = Column(String(250))
code = Column(String(100),unique=True)
expires_at = Column(DateTime, onupdate=datetime.datetime.now)
engine = create_engine('sqlite:///sqlalchemy_oauth.db')
Base.metadata.create_all(engine)
我的验证器看起来像这样
from oauthlib.oauth2 import RequestValidator
from modelsforoauth import User ,Client, Base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sqlalchemy_oauth.db', echo=True)
Base.metadata.bind = engine
DBSession = sessionmaker()
DBSession.bind = engine
session = DBSession()
#session.query(Client.client_id).first()
class MyRequestValidator(RequestValidator):
def validate_client_id(self, client_id, request):
try:
session.query(Client).filter_by(client_id=client_id).first()
return True
except Client.DoesNotExist:
return False
我想在 falcon 中为我的 api 定义端点。
from My_validator import MyRequestValidator
from oauthlib.oauth2 import WebApplicationServer
validator = MyRequestValidator()
server = WebApplicationServer(validator)
class AuthorizationView(object):
def __init__(self):
self._authorization_endpoint = server
def on_get(self, req, resp):
当我在下面查看文档时 link oauthlibserver
这清楚地说
uri, http_method, body, headers = extract_params(请求)
我如何在猎鹰中实现这一目标
我还想根据 documentaion
中的要求检查我的数据模型是否正确
我也不明白范围应该是什么 uri_redirect
如果我必须对模型或代码进行一些更改,请专家告诉我
组建新蜜蜂
提前致谢
您的提取参数函数应如下所示:
def extract_params(request):
# returns uri, http_method, body, headers
return request.uri, request.method, request.stream.read(), request.headers
我的数据库是这样的
from sqlalchemy.ext.declarative import declarative_base
import os
import time
import datetime
import sys
import uuid
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
from sqlalchemy import *
from sqlalchemy_utils.types.choice import ChoiceType
from sqlalchemy_utils.types.url import URLType
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
fullname = Column(String(250), nullable=False)
password = Column(String(250), nullable=False)
def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
class Client(Base):
Grand_choice = [('authorization_code', 'Authorization code')]
Response_choice = [('code', 'Authorization code')]
__tablename__ = 'client'
id = Column(Integer, primary_key=True)
client_id = Column(String(250), default=uuid.uuid4() ,unique=True)
user = Column(Integer, ForeignKey('user.id'))
grant_type = Column(ChoiceType(Grand_choice))
response_type = Column(ChoiceType(Response_choice))
scopes = Column(String(260))
default_scopes = Column(String(260))
redirect_uri = Column(URLType)
default_redirect_uri = Column(URLType)
class Bearer_Token(Base):
__tablename__ = 'Bearer_Token'
id = Column(Integer, primary_key=True)
client = Column(Integer, ForeignKey('client.id'))
user = Column(Integer, ForeignKey('user.id'))
scopes = Column(String(250))
access_token = Column(String(100),unique=True)
refresh_token = Column(String(100),unique=True)
expires_at = Column(DateTime, onupdate=datetime.datetime.now)
class Authorization_Code(Base):
__tablename__ = 'Authorization_code'
id = Column(Integer, primary_key=True)
client = Column(Integer, ForeignKey('client.id'))
user = Column(Integer, ForeignKey('user.id'))
scopes = Column(String(250))
code = Column(String(100),unique=True)
expires_at = Column(DateTime, onupdate=datetime.datetime.now)
engine = create_engine('sqlite:///sqlalchemy_oauth.db')
Base.metadata.create_all(engine)
我的验证器看起来像这样
from oauthlib.oauth2 import RequestValidator
from modelsforoauth import User ,Client, Base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine('sqlite:///sqlalchemy_oauth.db', echo=True)
Base.metadata.bind = engine
DBSession = sessionmaker()
DBSession.bind = engine
session = DBSession()
#session.query(Client.client_id).first()
class MyRequestValidator(RequestValidator):
def validate_client_id(self, client_id, request):
try:
session.query(Client).filter_by(client_id=client_id).first()
return True
except Client.DoesNotExist:
return False
我想在 falcon 中为我的 api 定义端点。
from My_validator import MyRequestValidator
from oauthlib.oauth2 import WebApplicationServer
validator = MyRequestValidator()
server = WebApplicationServer(validator)
class AuthorizationView(object):
def __init__(self):
self._authorization_endpoint = server
def on_get(self, req, resp):
当我在下面查看文档时 link oauthlibserver
这清楚地说 uri, http_method, body, headers = extract_params(请求)
我如何在猎鹰中实现这一目标 我还想根据 documentaion
中的要求检查我的数据模型是否正确我也不明白范围应该是什么 uri_redirect
如果我必须对模型或代码进行一些更改,请专家告诉我 组建新蜜蜂
提前致谢
您的提取参数函数应如下所示:
def extract_params(request):
# returns uri, http_method, body, headers
return request.uri, request.method, request.stream.read(), request.headers