文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

从头完成一个 Restful API 服务

2024-12-01 14:16

关注

一、框架概要

先来看看大致的代码框架。

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

二、代码详解

1,依赖包

放到了requirements文件里。

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文件里。

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 数据库,当然也可以使用其他关系型数据库,比如 MySQL 等。

3,程序入口文件

manage.py

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 并注册模块。

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里引入蓝图。

from flask import Blueprint

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

5,表模型

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

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 "".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 库。

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。

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()
api_user.add_resource(UserAddApi, '/useradd', endpoint='useradd')

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

三、初始化

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

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

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

到此,初始化工作完毕。

四、测试API

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

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

发送请求测试如下:

如果修改名字为错误值,都会鉴权失败。

来源:萝卜大杂烩内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯