flask_restful无法使用 flask.ext.login 的login_required装饰器

Question

所以我在flask中编写了一个非常基本的restful API，@login_required decorator抛出了"as_view"错误。我知道这是装饰者，因为如果我把它注释掉，应用程序就会运行得很好。任何想法/解决方案。我尝试在User类中继承UserMixin和Resource。我在某个地方读到它与资源视图被继承有关。如果重要的话，我使用flask脚本来运行应用程序。谢谢各位。代码下方，代码下的回溯

import json
from flask import Flask, request
from flask_restful import Resource, abort, Api, reqparse
from flask.views import MethodView
from flask.ext.login import LoginManager, UserMixin, login_required, login_user, logout_user, current_user
from lib.mongo import db
from bson.objectid import ObjectId
from crontab import CronTab

app = Flask(__name__)
app.config["SECRET_KEY"] = ''
login_manager = LoginManager()
login_manager.init_app(app)
api = Api(app)
##############################################
##############################################
#######USER CLASS FOR LOGGING IN AND OUT
##############################################
##############################################
class User(UserMixin):
def __init__(self, username, api_key, id, active=True):
self.username = username
self.api_key = api_key
self.id = id
self.active = active
def is_active(self):
account = db.users.find_one({ 'username': self.username, 'api_key': self.api_key})
if account is not None:
if not account['username'] == self.username and account['api_key'] == self.api_key:
self.active = False
else:
self.active = False
return self.active
def is_anonymous(self):
return False
def is_authenticated(self):
return True
def get_id(self):
return str(db.accounts.find_one({ 'username': self.username, 'api_key': self.api_key})['_id'])

@login_manager.user_loader
def load_user(userid):
user_rec = db.accounts.find_one({'_id': ObjectId(userid)})
user = User(user_rec['username'], user_rec['api_key'], user_rec['_id'])
return user

class Auth(Resource):
def get(self):
api_key = request.args.get('api_key', '')
username = request.args.get('username', '')
if username == '':
abort(400, message='username was not provided')
if api_key == '':
abort(400, message='api key was not provided')
account = db.accounts.find_one({'username': username, 'api_key': api_key})
if account:
if account['account_type'] == 'admin':
user = User(username, api_key, account['_id'])
login_user(user)
return {'auth': 'successful', 'status_code': 200}, 200
else:
abort(401, message='you are not an admin')
else:
abort(401, message='could not auth account')

@login_required
class ListCronJobs(Resource):
def get(self):
cron = CronTab(user=True)
cronjobs = []
for line in cron.lines:
cronjobs.append(line)
return {'data': cronjobs}

api.add_resource(Auth, '/api/auth')
api.add_resource(ListCronJobs, '/api/listcronjobs')
if __name__ == '__main__':
app.run(debug=True)

追溯：

File "manage.py", line 2, in <module>
from app import app
File "/Users/user/folder/app/__init__.py", line 90, in <module>
api.add_resource(ListCronJobs, '/ListCronJobs/listcronjobs')
File "/Users/user/folder/venv/lib/python2.7/site-packages/flask_restful/__init__.py", line 401, in add_resource
self._register_view(self.app, resource, *urls, **kwargs)
File "/Users/user/folder/venv/lib/python2.7/site-packages/flask_restful/__init__.py", line 441, in _register_view
resource_func = self.output(resource.as_view(endpoint, *resource_class_args,
AttributeError: 'function' object has no attribute 'as_view'

编辑：所以我通过在get(self)上方放置@login_required而不是装饰类来解决这个问题

class ListCronJobs(Resource):
@login_required
def get(self):
cron = CronTab(user=True)
cronjobs = []
for line in cron.lines:
cronjobs.append(line)
return {'data': cronjobs}

新的问题是，当我提出这个请求时(curl)http://127.0.0.1:5000/api/listcronjobs")我收到错误："消息"："服务器无法验证您是否有权访问请求的URL。您提供了错误的凭据(例如错误的密码)，或者您的浏览器不了解如何提供所需的凭据。">

我在使用curl 进行身份验证后提出请求

Answer 1

所以这是工作版本。

from flask import Flask, request
from flask_restful import Resource, abort, Api
from flask.ext.login import LoginManager, UserMixin, login_required, login_user#, logout_user, current_user
from lib.mongo import db
from bson.objectid import ObjectId
from crontab import CronTab

app = Flask(__name__)
app.config["SECRET_KEY"] = 'm\x88xe2@xb18\x82xa9x7f\x84l[Gx8dxe6Nx8fJl\x95{xd2F\x10'
login_manager = LoginManager()
login_manager.init_app(app)
api = Api(app)
##############################################
##############################################
#######USER CLASS FOR LOGGING IN AND OUT
##############################################
##############################################
class User(UserMixin):
def __init__(self, username, api_key, id, active=True):
self.username = username
self.api_key = api_key
self.id = id
self.active = active
def is_active(self):
account = db.users.find_one({ 'username': self.username, 'api_key': self.api_key})
if account is not None:
if not account['username'] == self.username and account['api_key'] == self.api_key:
self.active = False
else:
self.active = False
return self.active
def is_anonymous(self):
return False
def is_authenticated(self):
return True
def get_id(self):
return str(db.accounts.find_one({ 'username': self.username, 'api_key': self.api_key})['_id'])

@login_manager.user_loader
def load_user(userid):
user_rec = db.accounts.find_one({'_id': ObjectId(userid)})
user = User(user_rec['username'], user_rec['api_key'], user_rec['_id'])
return user

class Auth(Resource):
def get(self):
api_key = request.args.get('api_key', '')
username = request.args.get('username', '')
if username == '':
abort(400, message='username was not provided')
if api_key == '':
abort(400, message='api key was not provided')
account = db.accounts.find_one({'username': username, 'api_key': api_key})
if account:
if account['account_type'] == 'admin':
user = User(username, api_key, account['_id'])
login_user(user)
return {'auth': 'successful', 'status_code': 200}, 200
else:
abort(401, message='you are not an admin')
else:
abort(401, message='could not auth account')

class ListCronJobs(Resource):
@login_required
def get(self):
cron = CronTab(user=True)
cronjobs = []
for line in cron.lines:
cronjobs.append(line)
return {'data': cronjobs}
api.add_resource(Auth, '/api/auth')
api.add_resource(ListCronJobs, '/api/listcronjobs')
if __name__ == '__main__':
app.run(debug=True)

我没有用curl运行测试，而是用python的请求模块运行它们，并创建了一个会话。

import requests
s = requests.Session()
res = s.get('http://127.0.0.1:5000/api/auth?username=foo&api_key=bar')
print res.json()
res = s.get('http://127.0.0.1:5000/api/listcronjobs')
print res.json()

当在会话中使用curl进行身份验证时，没有保留，因此列出cronjobs的API调用被拒绝。