我在测试使用 Flask-Restplus 制作的 API 并将其集成到现有应用程序中时遇到了麻烦。
项目结构（不完整，相关部分）如下（根文件夹称为portal）
|
|-api -> Blueprint directory package
| |
| |-> endpoints -> package for endpoints
| | |-> __init__.py -> Empty
| | |-> token.py -> The token endpoint
| |-> __init__.py -> empty
| |-> business.py -> logic to create tokens
| |-> restplus.py -> API initialization
| |-> serializer.py -> token serializer
|-tests -> Tests folder
| |-> helpers.py -> Base classes for testing, setup, etc
| |-> tests_api.py -> API tests
|-> app.py -> app initialization and config
|-> main.py -> Entry point
主文件
import sys
from flask import Blueprint
from portal.app import app
from portal import libs
from portal.api.restplus import api
from portal.api.endpoints.token import ns as tokens_namespace
if __name__ == '__main__':
if len(sys.argv) == 2:
port = int(sys.argv[1])
else:
port = 5000
host = app.config.get('HOST', '127.0.0.1')
# Configure the Blueprint for API
blueprint = Blueprint('api', __name__, url_prefix='/api')
api.init_app(blueprint)
api.add_namespace(tokens_namespace)
app.register_blueprint(blueprint)
# For Dev environment we set the app root path to the current working directory
import os
app.root_path = os.getcwd()
print "Running in", app.root_path, " with DEBUG=", app.config.get('DEBUG', False)
app.run(host,
port,
app.config.get('DEBUG', False),
use_reloader=True
)
应用程序.py
import sys
import os
import datetime
import logging
from logging import Formatter
from logging.handlers import RotatingFileHandler
from jinja2 import Environment, PackageLoader
from flask import Flask, url_for, render_template, abort, request
from utils.mail import CustomMail as Mail
app = Flask(__name__)
app.secret_key = "crackthis"
app.config.from_pyfile('settings.py')
if os.path.exists(os.path.join(app.root_path, 'local_settings.py')):
app.config.from_pyfile('local_settings.py')
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgres://%s@%s/%s' % (
app.config['DB_USER'], app.config['DB_HOST'], app.config['DB_NAME'])
def in_test():
return "".join(sys.argv).find('nosetests') >= 0
Environment(
loader=PackageLoader('portal', 'templates')).globals['url_for'] = url_for
mail = Mail(app)
from filters import CUSTOM_FILTERS
app.jinja_env.filters.update(CUSTOM_FILTERS)
app.jinja_env.globals.update(zip=zip, now=datetime.datetime.now())
# Initialize event handling
from blinker import Namespace
signals = Namespace()
from portal.events import handlers
def log_exception(exc_info, app_ctx):
app.logger.exception(
"""
Request: {method} {path}
IP: {ip}
Agent: {agent_platform} | {agent_browser} {agent_browser_version}
Raw Agent: {agent}
user: {user}
""".format(
method=request.method,
path=request.path,
ip=request.remote_addr,
agent_platform=request.user_agent.platform,
agent_browser=request.user_agent.browser,
agent_browser_version=request.user_agent.version,
agent=request.user_agent.string,
user=app_ctx.logged_in
)
)
@app.errorhandler(500)
def server_error_page(error):
from portal.utils.app_ctx import AppContext
app_ctx = AppContext('connect', 'error')
log_exception(error, app_ctx)
return render_template("errors/server_error_new.html", app_ctx=app_ctx), 500
@app.errorhandler(404)
def server_page_not_found(error):
from portal.utils.app_ctx import AppContext
app_ctx = AppContext('connect', 'error')
log_exception(error, app_ctx)
return render_template("errors/page_not_found_new.html", app_ctx=app_ctx), 404
现在是蓝图部分
令牌.py
from flask import request
from flask_restplus import Resource
from portal.api.business import create_token, update_token, delete_token
from portal.api.serializers import token
from portal.api.restplus import api
from portal.utils.api_helpers import api_authenticate, is_administrator, is_assistant
from portal.models.api import ApiToken
ns = api.namespace('tokens', description='Operations related to deal with tokens')
@ns.route('/')
class TokenCollection(Resource):
parser = api.parser()
parser.add_argument('email', type=str, help='Member username', location='form')
parser.add_argument('password', type=str, help='Member password', location='form')
@api.response(201, 'Token successfully created.')
@api.response(404, 'Not user with provided credentials')
@api.response(403, 'Provided password or username are invalid')
@api.expect(parser)
def post(self):
"""
Creates a new token.
"""
data = request.form
response, code = create_token(data)
return response, code
@ns.route('/<string:id>')
@api.response(404, 'Token not found.')
class TokenItem(Resource):
method_decorators = [is_administrator, api_authenticate]
@api.marshal_with(token, code=200)
@api.header('Auth-Token', 'Required field for accessing most API endpoints', required=True)
def get(self, id):
"""
Returns a Token.
"""
token = ApiToken.query.filter(ApiToken.id == id).one()
return token
@api.response(204, 'Token successfully deleted.')
@api.header('Auth-Token', 'Required field for accessing most API endpoints', required=True)
def delete(self, id):
"""
Deletes a Token.
"""
delete_token(id)
return None, 204
业务.py
from datetime import datetime
from portal.db import DB
from portal.models import Program, Member
from portal.models.api import ApiToken
def create_token(data):
"""
Creates a token to be used in furhter API calls
:param data: A Dict containing username and password
:return:
"""
member = Member()
member.from_dict(data)
password = member.password
username = member.email
try:
member = Member.objects.get(username__iexact=username, skip_auth=True)
except Member.DoesNotExist:
result = {'error': 'Not user with provided credentials'}
code = 404
return result, code
if not member.check_password(password):
result = {'error': 'Provided password or username are invalid'}
code = 403
return result, code
try:
token = ApiToken.objects.get(member=member)
except ApiToken.DoesNotExist:
token = ApiToken(member=member)
token.save()
code = 201
return {'id': token.id,
'token': token.token,
'member_id': member.id,
'date': str(token.date)}, code
def update_token(token_id, data):
"""
Updates date and deleted flag of specified token
:param token_id: The token id
:param data: The dict with data
:return: The Token modifed
"""
try:
token = ApiToken.objects.get(id=token_id)
token.date = datetime.utcnow
token.is_deleted = data['is_deleted']
token.save()
except ApiToken.DoesNotExist:
result = {'error': 'Could not find Auth Token with provided ID'}
code = 404
return result, code
code = 204
return {'id': token.id,
'token': token.token,
'member_id': token.member_id,
'date': str(token.date)}, code
def delete_token(token_id):
"""
Deletes specified token if it exists
:param token_id: The token ID
:return:
"""
try:
ApiToken.objects.delete(id=token_id)
except ApiToken.DoesNotExist:
result = {'error': 'Could not find Auth Token with provided ID'}
code = 404
return result, code
code = 204
return 'Token deleted successfully', code
restplus.py
from flask_restplus import Api
from sqlalchemy.orm.exc import NoResultFound
api = Api(version='1.0', title='Wizbots API',
description='Wizbots site Restful API')
@api.errorhandler
def default_error_handler(e):
message = 'An unhandled exception occurred.'
return {'message': message}, 500
@api.errorhandler(NoResultFound)
def database_not_found_error_handler(e):
return {'message': 'A database result was required but none was found.'}, 404
序列化程序.py
from flask_restplus import fields
from portal.api.restplus import api
token = api.model('ApiToken', {
'id': fields.String(readOnly=True, description='The unique identifier of a Token'),
'token': fields.String(required=True, description='The token value'),
'member_id': fields.String(required=True, description='The Member ID associated with token'),
'date': fields.DateTime(required=True, description='Token creation time'),
})
现在测试本身：
测试api.py
"""
Test API endpoints
"""
import unittest
from portal.tests import helpers
from portal.app import app
from portal.views import (
views, account, admin, enrollment,
programs, lab, public)
class ApiTests(helpers.ViewBase):
############################
#### setup and teardown ####
############################
def setUp(self):
super(ApiTests, self).setUp()
app.config['TESTING'] = True
app.config['WTF_CSRF_ENABLED'] = False
app.config['DEBUG'] = False
self.assertEquals(app.debug, False)
# executed after to each test
def tearDown(self):
pass
########################
#### helper methods ####
########################
###############
#### tests ####
###############
def test_can_obtain_token(self):
# Make a call to api endpoint to create a token
form_data = dict()
form_data['email'] = 'admin@wizbots.test.com'
form_data['password'] = 'password'
response = self.client.post('/api/tokens', data=form_data)
# Make sure response is ok and contains and Auth-token
self.assertEqual(response.status_code, 200)
self.assertIn('token', response.data)
def test_wrong_credentials_token(self):
pass
if __name__ == "__main__":
unittest.main()
使用鼻子测试运行测试时：
nosetests -s tests/test_api.py:ApiTests.test_can_obtain_token
我继续得到404：
F
======================================================================
FAIL: test_can_obtain_token (portal.tests.test_api.ApiTests)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/internetmosquito/git/wizbots/portal/src/portal/tests/test_api.py", line 42, in test_can_obtain_token
form_data = dict()
AssertionError: 404 != 200
-------------------- >> begin captured logging << --------------------
portal.app: ERROR:
Request: POST /api/tokens
IP: None
Agent: None | None None
Raw Agent:
user: <portal.models.member.AnonymousMember instance at 0x7f73b4654560>
Traceback (most recent call last):
File "/home/internetmosquito/python_envs/wizbots/local/lib/python2.7/site-packages/flask/app.py", line 1639, in full_dispatch_request
rv = self.dispatch_request()
File "/home/internetmosquito/python_envs/wizbots/local/lib/python2.7/site-packages/flask/app.py", line 1617, in dispatch_request
self.raise_routing_exception(req)
File "/home/internetmosquito/python_envs/wizbots/local/lib/python2.7/site-packages/flask/app.py", line 1600, in raise_routing_exception
raise request.routing_exception
NotFound: 404: Not Found
--------------------- >> end captured logging << ---------------------
----------------------------------------------------------------------
Ran 1 test in 3.967s
看起来好像 /api/tokens 端点无法访问......不用说，如果我启动服务器并使用 curl 或 swaggerUI 的端点......知道我在做什么错了吗？谢谢！