0

我的数据库看起来像这样

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):

当我查看以下链接oauthlibserver中的文档时

这清楚地表明 uri, http_method, body, headers = extract_params(request)

我如何在 falcon 中实现这一点，我还想根据文档中的要求检查我的数据模型是否正确

我也不明白范围和 uri_redirect 应该是什么

如果我必须对模型或代码进行一些更改，请专家让我知道形成一个新手

提前致谢

4

1 回答 1

1

您的提取参数函数应如下所示：

def extract_params(request): # returns uri, http_method, body, headers return request.uri, request.method, request.stream.read(), request.headers

于 2017-02-07T22:45:11.463 回答