从头完成一个restful API 服务

“ 昨天搭建了一个HTTPS的服务器，放在那里不用简直是浪费了，今天来部署一个RESTFul API服务”

走起！

01.框架概要

先来看看大致的代码框架

先说明下，这套代码结构是参照经典flask书籍《Flask Web Development》来写的，想要这本书电子版的，可以联系我哈。

02.代码详解

1，依赖包

放到了requirements文件里

代码语言：javascript复制

flask==1.0.2
flask-script==2.0.6
flask-restful==0.3.7
flask-sqlalchemy==2.3.3
flask-httpauth==3.2.4
itsdangerous==1.1.0
Flask-SQLAlchemy==2.3.2
psycopg2==2.7.6.1

2，配置文件

在config.py文件里

代码语言：javascript复制

import os


basedir = os.path.abspath(os.path.dirname(__file__))


class Config:

    @staticmethod
    def init_app(app):
        pass
class DevelopmentConfig(Config):
    pass
class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///'   os.path.join(basedir, 'my.sqlite3')


class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')


config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

这里我使用了简单易用的sqlite数据库。

3，程序入口文件

manage.py

代码语言：javascript复制

from app import create_app, db
from flask_script import Manager, Shell, Server
from app.models import User, AdminUser, Picture


app = create_app('testing')
manager = Manager(app)


def make_shell_context():
    return dict(app=app, db=db, User=User, AdminUser=AdminUser, Picture=Picture)


manager.add_command("runserver", Server(use_debugger=True, host='0.0.0.0', port='9980'))
manager.add_command("shell", Shell(make_context=make_shell_context))


if __name__ == '__main__':
    manager.run(default_command='runserver')

引入了flask的command用法，后面主要用来做数据库的初始化操作和快速启动flask服务。

4，引入蓝图

蓝图可以帮我们实现模块化应用的功能。

在主程序的__init__.py文件中创建flask app并注册模块

代码语言：javascript复制

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import config

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 .main import main as main_blueprint
    app.register_blueprint(main_blueprint)
    from .api_1_0 import api_1_0 as api_blueprint
    app.register_blueprint(api_blueprint)

    return app

同时在api模块的__init__.py里引入蓝图

代码语言：javascript复制

from flask import Blueprint

api_1_0 = Blueprint('api_1_0', __name__, url_prefix='/api')
from . import api_user, api_auth

5，表模型

在models.py文件里，定义了当前用到的数据库表结构

代码语言：javascript复制

from . import db
from werkzeug.security import generate_password_hash, check_password_hash


class AdminUser(db.Model):
    __tablename__ = 'adminuser'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(128), unique=True, index=True)
    password_hash = db.Column(db.String(128))

    @staticmethod
    def init_user():
        users = AdminUser.query.filter_by(username='admin').first()
        if users is None:
            users = AdminUser(username='admin')
        users.password = 'hardtoguess'
        db.session.add(users)
        db.session.commit()

    @property
    def password(self):
        raise AttributeError('password is not readable attribute')

    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)

    def verify_password(self, password):
        return check_password_hash(self.password_hash, password)

    def __repr__(self):
        return "<Users {}>".format(self.username)


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(128), unique=True, index=True)
    picture_count = db.Column(db.Integer)


class Picture(db.Model):
    __tablename__ = 'pictures'
    id = db.Column(db.Integer, primary_key=True)
    picture_id = db.Column(db.Integer, db.ForeignKey('users.id'))
    picture_name = db.Column(db.String(128))
    picture = db.Column(db.Text)

其中AdminUser是作为API的鉴权用户的存在，后面调用API都会使用该用户，而User和Picture则是后面需要操作的表了。

着重说下AdminUser类，定义了一个静态方法init_user，是用来后面初始化数据库的，我们需要把这个鉴权用户手动添加到数据库中。方法verify_password用于后面API的鉴权，如果数据库存在该用户且密码的哈希值相同，则鉴权通过。

6，API部分

api鉴权代码，使用flask_httpauth库

代码语言：javascript复制

from flask_httpauth import HTTPBasicAuth
from flask import jsonify
from ..models import AdminUser


auth = HTTPBasicAuth()


@auth.error_handler
def unauthorized():
    error_info = '{}'.format("Invalid credentials")
    print(error_info)
    response = jsonify({'error': error_info})
    response.status_code = 403
    return response


@auth.verify_password
def verify_password(username, password):
    user = AdminUser.query.filter_by(username=username).first()
    if not user or not user.verify_password(password):
        return False
    return True

先到数据库查询用户，如果不存在或者密码哈希值不正确，则返回False，鉴权失败。

7，设计API

在api_user.py文件中，初始化flask_restful的Api类，用于后面添加resource

代码语言：javascript复制

api_user = Api(api_1_0)


class UserAddApi(Resource):
    @auth.login_required
    def post(self):
        user_info = request.get_json()
        try:
            u = User(username=user_info['username'])
            p = Picture(picture_name=user_info['username'])
            u.password = user_info['password']
            u.picture_count = user_info['picture_count']
            db.session.add(u)
            db.session.add(p)
            db.session.commit()
        except:
            logger.info("User add: {} failure...".format(user_info['username']))
            db.session.rollback()
            return False
        else:
            logger.info("User add: {} success...".format(user_info['username']))
            return True
        finally:
            db.session.close()

代码语言：javascript复制

api_user.add_resource(UserAddApi, '/useradd', endpoint='useradd')

这里只接受json的请求消息体，并且对应的更新数据库中表的值。

03.初始化

使用在manage中定义的shell命令，进入shell界面

执行db.create_all()创建在models中定义的表，执行成功之后，查看数据库如下

表已经创建成功，但是我们的admin用户还不存在，继续执行命令AdminUser.init_user()，插入Admin用户。

到此，初始化工作完毕

04.测试API

执行命令python manage.py启动flask服务，可以看到已经在本地的9980端口启动

使用postman工具测试useraddAPI，选择POST方式，鉴权方式使用Basic Auth，body中使用json形式

发送请求测试如下：

如果修改名字或者密码为错误值，都会鉴权失败

04. 部署到服务器

接下来和我们前面搭建的HTTPS服务器结合到一起

把代码包上传到服务器，然后重新做一遍初始化操作，修改run.sh脚本到manage，启动，测试，完成！

python 数据库 sql flask api

0 人点赞