Flask开发接口

# coding=utf-8
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import config

import sys
reload(sys)
sys.setdefaultencoding(‘utf-8‘)


db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    db.init_app(app)


    from .api_1_0 import api as api_1_0_blueprint
    app.register_blueprint(api_1_0_blueprint,url_prefix=‘/api/v1.0‘)

    return app

# coding=utf-8
from datetime import datetime
from . import db
from app.exception import ValidationError
from flask import url_for


class Post(db.Model):
    __tablename__=‘posts‘
    id = db.Column(db.Integer,primary_key=True)
    body = db.Column(db.Text)
    timestamp = db.Column(db.DateTime,index=True,default=datetime.utcnow)

# 所有 url_for() 方法都指定了参数 _external=True，
# 这么做是为了生成完整的 URL，而不是生成传统 Web 程序中经常使用的
# 相对 URL。
    def to_json(self):
        json_post = {
            ‘url‘:url_for(‘api.get_post‘,id=self.id,_external=True),
            ‘body‘:self.body,
            ‘timestamp‘:self.timestamp
        }
        return json_post

# 因为id,和timestamp都是自动生成的，所以只需要传body的值即可
    @staticmethod
    def from_json(json_post):
        body = json_post.get(‘body‘)
        if body is None or body ==‘‘:
            raise ValidationError(u‘文章内容不能为空‘)
        return Post(body=body)

from flask import Blueprint

api = Blueprint(‘api‘,__name__)

from . import posts

# coding=utf-8
from flask import jsonify
from . import api
from app.exception import  ValidationError
import sys
reload(sys)
sys.setdefaultencoding(‘utf-8‘)



def bad_request(message):
    response = jsonify({‘错误‘:‘请求错误‘,‘信息‘:message})
    response.status_code = 400
    return response


# 程序需要向客户端提供适当的响应以处理这个异常。为了避免在视图函数中编写捕
# 获异常的代码， 我们可创建一个全局异常处理程序。
@api.errorhandler(ValidationError)
def validation_error(e):
    return bad_request(e.args[0])

# coding=utf-8
from flask import jsonify, request, g, url_for

from app.models import Post
from . import api
from .. import db



# 获取文章资源GET请求
@api.route(‘/posts/<int:id>‘)
def get_post(id):
    post = Post.query.get_or_404(id)
    return jsonify(post.to_json())


# 文章资源POST 插入请求
@api.route(‘/posts‘, methods=[‘POST‘])
def new_post():
    post = Post.from_json(request.json)
    db.session.add(post)
    db.session.commit()
    return jsonify(post.to_json()), 201,{‘Location‘: url_for(‘api.get_post‘, id=post.id, _external=True)}

# coding=utf-8
import os
basedir = os.path.abspath(os.path.dirname(__file__))

class Config:

    @staticmethod
    def init_app(app):
        pass


class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = ‘sqlite:///D://py//flask-api//data-dev.sqlite‘



config = {
    ‘development‘: DevelopmentConfig,
    ‘default‘: DevelopmentConfig
}

# coding=utf-8
import os
from app import create_app, db
from app.models import Post
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, Post=Post)

manager.add_command("shell", Shell(make_context=make_shell_context))
manager.add_command(‘db‘, MigrateCommand)


@app.route(‘/‘)
def hello():
    return ‘hello‘


if __name__ == ‘__main__‘:
    manager.run()