python flask web 博客实例 restful api 5

2019-07-05 11:09:45 浏览数 (1)

|-flasky |-app/ |-api_1_0 |-init.py |-users.py |-posts.py |-comments.py |-authentication.py |-errors.py |-decorators.py

1 app/api_1_0/init.py from flask import Blueprint api = Blueprint('api', name) from . import authentication, posts, users, comments, errors

2 app/init.py def create_app(config_name): # ... from .api_1_0 import api as api_1_0_blueprint app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1.0')

3 app/main/errors.py @main.app_errorhandler(404) def page_not_found(e): if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html: response = jsonify({'error': 'not found'}) response.status_code = 404 return response return render_template('404.html'), 404

4 　app/api_1_0/errors.py def forbidden(message): response = jsonify({'error': 'forbidden', 'message': message}) response.status_code = 403 return response

5 pip install flask-httpauth #验证 6 app/api_1_0/authentication.py from flask.ext.httpauth import HTTPBasicAuth from .errors import forbidden_error auth = HTTPBasicAuth() @auth.verify_password def verify_password(email, password): if email == '': g.current_user = AnonymousUser() return True user = User.query.filter_by(email = email).first() if not user: return False g.current_user = user return user.verify_password(password) @auth.error_handler def auth_error(): return unauthorized('Invalid credentials') @api.before_request @auth.login_required def before_request(): if not g.current_user.is_anonymous and not g.current_user.confirmed: return forbidden('Unconfirmed account') @auth.verify_password def verify_password(email_or_token, password): if email_or_token == '': g.current_user = AnonymousUser() return True if password == '': g.current_user = User.verify_auth_token(email_or_token) g.token_used = True return g.current_user is not None user = User.query.filter_by(email=email_or_token).first() if not user: return False g.current_user = user g.token_used = False return user.verify_password(password) @api.route('/token') def get_token(): if g.current_user.is_anonymous() or g.token_used: return unauthorized('Invalid credentials') return jsonify({'token': g.current_user.generate_auth_token(expiration=3600),'expiration': 3600}) 7 app/models.py from app.exceptions import ValidationError class User(db.Model):

...

代码语言：javascript复制
def generate_auth_token(self, expiration):
    s = Serializer(current_app.config['SECRET_KEY'],expires_in=expiration)
    return s.dumps({'id': self.id})
@staticmethod
def verify_auth_token(token):
    s = Serializer(current_app.config['SECRET_KEY'])
    try:
        data = s.loads(token)
    except:
        return None
    return User.query.get(data['id'])
def to_json(self):
    json_user = {
          'url': url_for('api.get_post', id=self.id, _external=True),
          'username': self.username,
          'member_since': self.member_since,
          'last_seen': self.last_seen,
          'posts': url_for('api.get_user_posts', id=self.id, _external=True),
          'followed_posts': url_for('api.get_user_followed_posts',id=self.id,_external=True),
          'post_count': self.posts.count()
     }
    return json_user

class Post(db.Model):

...

代码语言：javascript复制
def to_json(self):
    json_post = {
          'url': url_for('api.get_post', id=self.id, _external=True),
          'body': self.body,
          'body_html': self.body_html,
          'timestamp': self.timestamp,
          'author': url_for('api.get_user', id=self.author_id,_external=True),
          'comments': url_for('api.get_post_comments', id=self.id,_external=True)
          'comment_count': self.comments.count()
      }
    return json_post
@staticmethod
def from_json(json_post):
    body = json_post.get('body')
    if body is None or body == '':
        raise ValidationError('post does not have a body')
    return Post(body=body)

7 　app/exceptions.py class ValidationError(ValueError): pass

8 app/api_1_0/errors.py @api.errorhandler(ValidationError) def validation_error(e): return bad_request(e.args[0])

9 app/api_1_0/posts.py @api.route('/posts/') @auth.login_required def get_posts(): posts = Post.query.all() return jsonify({ 'posts': [post.to_json() for post in posts] }) @api.route('/posts/<int:id>') @auth.login_required def get_post(id): post = Post.query.get_or_404(id) return jsonify(post.to_json()) @api.route('/posts/', methods=['POST']) @permission_required(Permission.WRITE_ARTICLES) def new_post(): post = Post.from_json(request.json) post.author = g.current_user db.session.add(post) db.session.commit() return jsonify(post.to_json()), 201, {'Location': url_for('api.get_post', id=post.id, _external=True)} @api.route('/posts/<int:id>', methods=['PUT']) @permission_required(Permission.WRITE_ARTICLES) def edit_post(id): post = Post.query.get_or_404(id) if g.current_user != post.author and not g.current_user.can(Permission.ADMINISTER): return forbidden('Insufficient permissions') post.body = request.json.get('body', post.body) db.session.add(post) return jsonify(post.to_json()) @api.route('/posts/') def get_posts(): page = request.args.get('page', 1, type=int) pagination = Post.query.paginate(page, per_page=current_app.config['FLASKY_POSTS_PER_PAGE'],error_out=False) posts = pagination.items prev = None if pagination.has_prev: prev = url_for('api.get_posts', page=page-1, _external=True) next = None if pagination.has_next: next = url_for('api.get_posts', page=page 1, _external=True) return jsonify({ 'posts': [post.to_json() for post in posts], 'prev': prev, 'next': next, 'count': pagination.total }

10 　app/api_1_0/decorators.py def permission_required(permission): def decorator(f): @wraps(f) def decorated_function(*args, *kwargs): if not g.current_user.can(permission): return forbidden('Insufficient permissions') return f(args, **kwargs) return decorated_function return decorator

11 app/main/views.py #关闭服务器路由 @main.route('/shutdown') def server_shutdown(): if not current_app.testing: abort(404) shutdown = request.environ.get('werkzeug.server.shutdown') if not shutdown: abort(500) shutdown() return 'Shutting down...' 12 app/main/views.py ##报告缓慢的数据库查询 from flask.ext.sqlalchemy import get_debug_queries @main.after_app_request def after_request(response): for query in get_debug_queries(): if query.duration >= current_app.config['FLASKY_SLOW_DB_QUERY_TIME']: current_app.logger.warning( 'Slow query: %snParameters: %snDuration: %fsnContext: %sn' % (query.statement, query.parameters, query.duration, query.context)) return response

13 manage.py #分析源码 @manager.command def profile(length=25, profile_dir=None): """Start the application under the code profiler.""" from werkzeug.contrib.profiler import ProfilerMiddleware app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[length],profile_dir=profile_dir) app.run()

14 app/email.py from threading import Thread from flask import current_app, render_template from flask_mail import Message from . import mail def send_async_email(app, msg): with app.app_context(): mail.send(msg) def send_email(to, subject, template, **kwargs): app = current_app._get_current_object() msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] ' ' subject, sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to]) msg.body = render_template(template '.txt', **kwargs) msg.html = render_template(template '.html', **kwargs) thr = Thread(target=send_async_email, args=[app, msg]) thr.start() return thr

15 manage.py

!/usr/bin/env python

import os COV = None if os.environ.get('FLASK_COVERAGE'): import coverage COV = coverage.coverage(branch=True, include='app/*') COV.start()

if os.path.exists('.env'): print('Importing environment from .env...') for line in open('.env'): var = line.strip().split('=') if len(var) == 2: os.environ[var[0]] = var[1]

from app import create_app, db from app.models import User, Follow, Role, Permission, Post, Comment from flask_script import Manager, Shell from flask_migrate import Migrate, MigrateCommand

app = create_app(os.getenv('FLASK_CONFIG') or 'default') manager = Manager(app) migrate = Migrate(app, db)

def make_shell_context(): return dict(app=app, db=db, User=User, Follow=Follow, Role=Role, Permission=Permission, Post=Post, Comment=Comment) manager.add_command("shell", Shell(make_context=make_shell_context)) manager.add_command('db', MigrateCommand)

@manager.command def test(coverage=False): """Run the unit tests.""" if coverage and not os.environ.get('FLASK_COVERAGE'): import sys os.environ['FLASK_COVERAGE'] = '1' os.execvp(sys.executable, [sys.executable] sys.argv) import unittest tests = unittest.TestLoader().discover('tests') unittest.TextTestRunner(verbosity=2).run(tests) if COV: COV.stop() COV.save() print('Coverage Summary:') COV.report() basedir = os.path.abspath(os.path.dirname(file)) covdir = os.path.join(basedir, 'tmp/coverage') COV.html_report(directory=covdir) print('HTML version: file://%s/index.html' % covdir) COV.erase()

@manager.command def profile(length=25, profile_dir=None): """Start the application under the code profiler.""" from werkzeug.contrib.profiler import ProfilerMiddleware app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[length], profile_dir=profile_dir) app.run()

@manager.command def deploy(): """Run deployment tasks.""" from flask_migrate import upgrade from app.models import Role, User

代码语言：javascript复制
# migrate database to latest revision
upgrade()

# create user roles
Role.insert_roles()

# create self-follows for all users
User.add_self_follows()

if name == 'main': manager.run()

16 config.py import os basedir = os.path.abspath(os.path.dirname(file))

class Config: SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string' SSL_DISABLE = False SQLALCHEMY_COMMIT_ON_TEARDOWN = True SQLALCHEMY_TRACK_MODIFICATIONS = False SQLALCHEMY_RECORD_QUERIES = True MAIL_SERVER = 'smtp.googlemail.com' MAIL_PORT = 587 MAIL_USE_TLS = True MAIL_USERNAME = os.environ.get('MAIL_USERNAME') MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD') FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]' FLASKY_MAIL_SENDER = 'Flasky Admin flasky@example.com' FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN') FLASKY_POSTS_PER_PAGE = 20 FLASKY_FOLLOWERS_PER_PAGE = 50 FLASKY_COMMENTS_PER_PAGE = 30 FLASKY_SLOW_DB_QUERY_TIME=0.5

代码语言：javascript复制
@staticmethod
def init_app(app):
    pass

class DevelopmentConfig(Config): DEBUG = True SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or 'sqlite:///' os.path.join(basedir, 'data-dev.sqlite')

class TestingConfig(Config): TESTING = True SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or 'sqlite:///' os.path.join(basedir, 'data-test.sqlite') WTF_CSRF_ENABLED = False

class ProductionConfig(Config): SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///' os.path.join(basedir, 'data.sqlite')

代码语言：javascript复制
@classmethod
def init_app(cls, app):
    Config.init_app(app)

    # email errors to the administrators
    import logging
    from logging.handlers import SMTPHandler
    credentials = None
    secure = None
    if getattr(cls, 'MAIL_USERNAME', None) is not None:
        credentials = (cls.MAIL_USERNAME, cls.MAIL_PASSWORD)
        if getattr(cls, 'MAIL_USE_TLS', None):
            secure = ()
    mail_handler = SMTPHandler(
        mailhost=(cls.MAIL_SERVER, cls.MAIL_PORT),
        fromaddr=cls.FLASKY_MAIL_SENDER,
        toaddrs=[cls.FLASKY_ADMIN],
        subject=cls.FLASKY_MAIL_SUBJECT_PREFIX   ' Application Error',
        credentials=credentials,
        secure=secure)
    mail_handler.setLevel(logging.ERROR)
    app.logger.addHandler(mail_handler)

class HerokuConfig(ProductionConfig): SSL_DISABLE = bool(os.environ.get('SSL_DISABLE'))

代码语言：javascript复制
@classmethod
def init_app(cls, app):
    ProductionConfig.init_app(app)

    # handle proxy server headers
    from werkzeug.contrib.fixers import ProxyFix
    app.wsgi_app = ProxyFix(app.wsgi_app)

    # log to stderr
    import logging
    from logging import StreamHandler
    file_handler = StreamHandler()
    file_handler.setLevel(logging.WARNING)
    app.logger.addHandler(file_handler)

class UnixConfig(ProductionConfig): @classmethod def init_app(cls, app): ProductionConfig.init_app(app)

代码语言：javascript复制
    # log to syslog
    import logging
    from logging.handlers import SysLogHandler
    syslog_handler = SysLogHandler()
    syslog_handler.setLevel(logging.WARNING)
    app.logger.addHandler(syslog_handler)

config = { 'development': DevelopmentConfig, 'testing': TestingConfig, 'production': ProductionConfig, 'heroku': HerokuConfig, 'unix': UnixConfig,

代码语言：javascript复制
'default': DevelopmentConfig

}

python manage.py profile 分析源码

https://github.com/miguelgrinberg/flasky

image.png

pip install httpie ##测试web服务

(venv) $ http --json --auth <email>:<password> GET

http://127.0.0.1:5000/api/v1.0/posts HTTP/1.0 200 OK Content-Length: 7018 Content-Type: application/json Date: Sun, 22 Dec 2013 08:11:24 GMT Server: Werkzeug/0.9.4 Python/2.7.3 { "posts": [ ... ], "prev": null "next": "http://127.0.0.1:5000/api/v1.0/posts/?page=2", "count": 150 }

(venv) $ http --json --auth : GET http://127.0.0.1:5000/api/v1.0/posts/

(venv) $ http --auth <email>:<password> --json POST

http://127.0.0.1:5000/api/v1.0/posts/ "body=I'm adding a post from the command line." HTTP/1.0 201 CREATED Content-Length: 360 Content-Type: application/json Date: Sun, 22 Dec 2013 08:30:27 GMT Location: http://127.0.0.1:5000/api/v1.0/posts/111 Server: Werkzeug/0.9.4 Python/2.7.3 { "author": "http://127.0.0.1:5000/api/v1.0/users/1", "body": "I'm adding a post from the command line.", "body_html": "<p>I'm adding a post from the <em>command line</em>.</p>", "comments": "http://127.0.0.1:5000/api/v1.0/posts/111/comments", "comment_count": 0, "timestamp": "Sun, 22 Dec 2013 08:30:27 GMT", "url": "http://127.0.0.1:5000/api/v1.0/posts/111" }

(venv) $ http --auth <email>:<password> --json GET

http://127.0.0.1:5000/api/v1.0/token HTTP/1.0 200 OK Content-Length: 162 Content-Type: application/json Date: Sat, 04 Jan 2014 08:38:47 GMT Server: Werkzeug/0.9.4 Python/3.3.3 { "expiration": 3600, "token": "eyJpYXQiOjEzODg4MjQ3MjcsImV4cCI6MTM4ODgyODMyNywiYWxnIjoiSFMy..." }

(venv) $ http --json --auth eyJpYXQ...: GET http://127.0.0.1:5000/api/v1.0/posts/

python json api http sqlite

0 人点赞