flask开发restful api

Posted 2020-07-02 Leo_wlCnBlogs

　在此之前，向大家说明的是，我们整个框架用的是flask + sqlalchemy + redis。如果没有开发过web，还是先去学习一下，这边只是介绍如果从开发web转换到开发移动端。如果flask还不是很熟悉，我建议先到这个网站简单学习一下，非常非常简单。http://dormousehole.readthedocs.org/en/latest/ 

　　一直想写一些特别的东西，能让大家学习讨论的东西。但目前网上的很多博客，老么就按照官方文档照本宣读，要么直接搬代码，什么都不说明。我写这个系列的博客，让大家由浅入深，一步一步走向复杂结构，以及为啥要这么走，其他方式可不可以等等。

　　目前看来，移动开发最火，而我们python最适合开发移动的就是flask web框架，这款web框架非常清晰，可以简单用，可以复杂用。最简单的时候，一个py文件，就可以做一个项目；复杂的时候，利用蓝图，做各种版本控制，代码结构自己完全控制，非常自由。

　　首先，我们要知道，目前我们移动开发基本都在用restful api，什么是restful api呢？百度百科一下：Web 应用程序最重要的 REST 原则是，客户端和服务器之间的交互在请求之间是无状态的。从客户端到服务器的每个请求都必须包含理解请求所必需的信息。如果服务器在请求之间的任何时间点重启，客户端不会得到通知。此外，无状态请求可以由任何可用服务器回答，这十分适合云计算之类的环境。客户端可以缓存数据以改进性能。

　　说白了我们不能使用cookie，不能使用session了。如果稍微有点http经验的人，都知道，很多时候，我们都把一些基本内容放在cookie里面，服务器每次读取或者写入的时候，服务器端就直接设置session就可以了，这样，每次，客户端直接携带自己的cookie值上来，我们就知道它是谁，怎么把数据给它。但restful api的风格不允许这样，那服务器应该采取何种方案呢？

　　目前网上大多数做法是token方式，第一次登录的时候，先提交用户名密码，服务器收集到以后，先验证一下，如果验证通过了，这时候服务器端基于用户名、密码、当前时间戳等内容，用md5或者des或者aes等加密方式，生成一个token值，然后把token值存放到redis里面，记录它对应哪个用户，然后把这个token值发给客户端。客户端收到token值以后，下次访问服务器端任何接口的时候，直接携带这个token，服务器端就知道它是谁了，该给它什么数据。哪以何种方式给服务器端呢？通常的做法就是把token值放在header里面。当然这个不是绝对，你也可以放在body里面，这个都是仁者见仁智者见智的事。好了，我们就先放在header头里面。

　　以上说了这么多，大家也烦了，直接上代码，再解释吧。

　　首先，创建整个工程，为了最简单，方便，我们直接就保留以下的py文件。

　　 

　　看model.py里面的代码：

　　

# coding:utf-8
from sqlalchemy import create_engine, ForeignKey, Column, Integer, String, Text, DateTime,\\
    and_, or_, SmallInteger, Float, DECIMAL, desc, asc, Table, join, event
from sqlalchemy.orm import relationship, backref, sessionmaker, scoped_session, aliased, mapper
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
from sqlalchemy.orm.collections import attribute_mapped_collection
import datetime

engine = create_engine("mysql://root:a12345678@127.0.0.1:3306/blog01?charset=utf8", pool_recycle=7200)

Base = declarative_base()

db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

Base.query = db_session.query_property()


class User(Base):
    __tablename__ = \'user\'

    id = Column(\'id\', Integer, primary_key=True)
    phone_number = Column(\'phone_number\', String(11), index=True)
    password = Column(\'password\', String(30))
    nickname = Column(\'nickname\', String(30), index=True, nullable=True)
    register_time = Column(\'register_time\', DateTime, index=True, default=datetime.datetime.now)


if __name__ == \'__main__\':
    Base.metadata.create_all(engine)

运行一下，就创建user表了。

user表中，有电话号码，phone_number；密码，password；昵称，nickname；register_time，注册时间

　　下面是view.py代码

 1 # coding:utf-8
 2 from flask import Flask, request, jsonify
 3 from model import User, db_session
 4 import hashlib
 5 import time
 6 import redis
 7 
 8 app = Flask(__name__)
 9 redis_store = redis.Redis(host=\'localhost\', port=6380, db=4, password=\'dahai123\')
10 
11 
12 @app.route(\'/\')
13 def hello_world():
14     return \'Hello World!\'
15 
16 
17 @app.route(\'/login\', methods=[\'POST\'])
18 def login():
19     phone_number = request.get_json().get(\'phone_number\')
20     password = request.get_json().get(\'password\')
21     user = User.query.filter_by(phone_number=phone_number).first()
22     if not user:
23         return jsonify({\'code\': 0, \'message\': \'没有此用户\'})
24 
25     if user.password != password:
26         return jsonify({\'code\': 0, \'message\': \'密码错误\'})
27 
28     m = hashlib.md5()
29     m.update(phone_number)
30     m.update(password)
31     m.update(str(int(time.time())))
32     token = m.hexdigest()
33 
34     redis_store.hmset(\'user:%s\' % user.phone_number, {\'token\': token, \'nickname\': user.nickname, \'app_online\': 1})
35     redis_store.set(\'token:%s\' % token, user.phone_number)
36     redis_store.expire(\'token:%s\' % token, 3600*24*30)
37 
38     return jsonify({\'code\': 1, \'message\': \'成功登录\', \'nickname\': user.nickname, \'token\': token})
39 
40 
41 @app.route(\'/user\')
42 def user():
43     token = request.headers.get(\'token\')
44     if not token:
45         return jsonify({\'code\': 0, \'message\': \'需要验证\'})
46     phone_number = redis_store.get(\'token:%s\' % token)
47     if not phone_number or token != redis_store.hget(\'user:%s\' % phone_number, \'token\'):
48         return jsonify({\'code\': 2, \'message\': \'验证信息错误\'})
49 
50     nickname = redis_store.hget(\'user:%s\' % phone_number, \'nickname\')
51     return jsonify({\'code\': 1, \'nickname\': nickname, \'phone_number\': phone_number})
52 
53 
54 @app.route(\'/logout\')
55 def logout():
56     token = request.headers.get(\'token\')
57     if not token:
58         return jsonify({\'code\': 0, \'message\': \'需要验证\'})
59     phone_number = redis_store.get(\'token:%s\' % token)
60     if not phone_number or token != redis_store.hget(\'user:%s\' % phone_number, \'token\'):
61         return jsonify({\'code\': 2, \'message\': \'验证信息错误\'})
62 
63     redis_store.delete(\'token:%s\' % token)
64     redis_store.hmset(\'user:%s\' % phone_number, {\'app_online\': 0})
65     return jsonify({\'code\': 1, \'message\': \'成功注销\'})
66 
67 
68 @app.teardown_request
69 def handle_teardown_request(exception):
70     db_session.remove()
71 
72 if __name__ == \'__main__\':
73     app.run(debug=True, host=\'0.0.0.0\', port=5001)

 

下面来逐个解释一下，

首先，几个import不用解释了，注意把User和db_session 都import过来，然后定义一个redis的connection。

再接下来就login接口，直接post方法，获取json格式数据，里面有phone_number和password，接下来，数据库查询，如果没有这个用户，返回一个json格式，再接下来对比password，如果phone_number和password都正确，就用md5函数，生成一个token，这个token包含由phone_number、password、当前时间戳str(int(time.time()))生成，再返回。

看我每个返回的东西，首先，都是json格式，这个是目前大多数移动开发默认返回的格式；其次，每个返回，必定有个code，目前这边有2个值，是0和1，其实可以看出来，0代表失败，1代表成功，有0的地方必定要有message。每个返回一个code是必须的，但是值，可以自己定义。很多开发把http的code直接拿来用，也可以，比如成功返回，就200，没有就404，禁止就403。这些都可以，只要服务器端开发和客户端开发开始就约定一个值就好，具体的值，只要能快速开发，都可以。

好了，返回格式先解释到这，我们以后会继续扩展。再看接下来的redis。

第一行，先用 user:13765505223 这种类型的作为每个用户的key，值是一些基本的东西，其中app_online，代表上线了，这个app_online，其实很重要的，因为移动端开发跟web不同，要记录移动端在登录状态，还是登出状态。如果在登录状态，我们就可以从服务器推送了，关于推送，我们以后会逐步讲，这个先放在这边。

第二行，用token:token 作为key，key里面的值是具体的用户电话号码

第三行，把这个token设置一个过期时间，超过这个时间，就删除，这个有需要的app可以设置一下。如果你的app的token想永远不变，这行代码可以注释掉。

好了，目前login函数基本完成。

接下来看验证函数user，和注销函数logout，

user这个函数，这是来验证登录以后，有没有数据，没多少意义。

逐行分析，首先在header里找到这个token，如果没有token，就返回失败；其次，验证redis，如果token所在的key value对里面没有值或者值错误，则返回失败。然后返回具体的数据。非常简单的一个函数。

下面的logout函数也差不多，也是这样，也是验证，然后删除token的key value对，再设置app_online为0，表示当前是注销状态。

最后一个很重要，这边一定要记住，把这个函数写上。如果没有这个函数，每一个会话以后，db_session都不会清除，很多时候，数据库改变了，前台找不到，或者明明已经提交，数据库还是没有更改，或者长时间没有访问接口，mysql gong away，这样的错误。总之，一定要加上。

好了，整个过程已经完成，下面进入验证状态。首先我们在外面新建一个用户，存到数据库，然后写个小脚本验证一下。

>>> from model import User, db_session
>>> new_user = User(phone_number=\'12345678901\', password=\'123456\', nickname=u\'测试用户1\')
>>> db_session.add(new_user)
>>> db_session.commit()

一个用户已经创建好，接下来就是测试，这边测试有2种方式，一个用IDE自带的测试软件测试，pycharm有很好的测试软件；其次用小脚本方式测试，既然我们以后要不停的写例子，就用小脚本测试吧，过程也非常简单。

 1 # coding:utf-8
 2 import requests
 3 import json
 4 
 5 
 6 class APITest(object):
 7     def __init__(self, base_url):
 8         self.base_url = base_url
 9         self.headers = {}
10         self.token = None
11 
12     def login(self, phone_number, password, path=\'/login\'):
13         payload = {\'phone_number\': phone_number, \'password\': password}
14         self.headers = {\'content-type\': \'application/json\'}
15         response = requests.post(url=self.base_url + path, data=json.dumps(payload), headers=self.headers)
16         response_data = json.loads(response.content)
17         self.token = response_data.get(\'token\')
18         return response_data
19 
20     def user(self, path=\'/user\'):
21         self.headers = {\'token\': self.token}
22         response = requests.get(url=self.base_url + path, headers=self.headers)
23         response_data = json.loads(response.content)
24         return response_data
25 
26     def logout(self, path=\'/logout\'):
27         self.headers = {\'token\': self.token}
28         response = requests.get(url=self.base_url + path, headers=self.headers)
29         response_data = json.loads(response.content)
30         return response_data

 

写一个很简单的小脚本，就可以拉到命令行测试了，我们试试吧。

>>> from client import APITest
>>> api = APITest(\'http://127.0.0.1:5001\')
>>> data = api.login(\'12345678901\', \'1234567\')
>>> print data.get(\'message\')
密码错误
>>> data = api.login(\'12345678901\', \'123456\')
>>> print data.get(\'message\')
成功登录
>>> data = api.user()
>>> print data
{u\'phone_number\': u\'12345678901\', u\'code\': 1, u\'nickname\': u\'\\u6d4b\\u8bd5\\u7528\\u62371\'}
>>> print nickname
Traceback (most recent call last):
  File "<input>", line 1, in <module>
NameError: name \'nickname\' is not defined
>>> print data.get(\'nickname\')
测试用户1
>>> data = api.logout()
>>> print data
{u\'message\': u\'\\u6210\\u529f\\u6ce8\\u9500\', u\'code\': 1}
>>> print message
Traceback (most recent call last):
  File "<input>", line 1, in <module>
NameError: name \'message\' is not defined
>>> print data.get(\'message\')
成功注销

 登录成功的时候，我们进redis看看redis数据格式，比较直观点。

127.0.0.1:6380[4]> keys *
1) "token:bbf73ab651a13a5bc5601cf01add2564"
2) "user:12345678901"
127.0.0.1:6380[4]> hgetall user:12345678901
1) "token"
2) "bbf73ab651a13a5bc5601cf01add2564"
3) "nickname"
4) "\\xe6\\xb5\\x8b\\xe8\\xaf\\x95\\xe7\\x94\\xa8\\xe6\\x88\\xb71"
5) "app_online"
6) "1"
127.0.0.1:6380[4]> get token:bbf73ab651a13a5bc5601cf01add2564
"12345678901"
127.0.0.1:6380[4]> 

 

嗯，一切都正常，但我们开发不能一切正常，就满足。就这些代码，我们有很多需要改进的地方，尤其是验证token的那边，是不是可以改进呢？redis设置的时候，一连串动作，如果这时候出错，或者redis只设置到一半怎么办？这些问题，我们下一章继续解决。

flask开发restful api系列(3)

摘要: 介绍了如何使用alembic来修改数据库阅读全文

posted @ 2016-03-31 08:53 月儿弯弯0204 阅读(259) | 评论 (0) 编辑

 

flask开发restful api系列(2)

摘要: 把上一章代码简化了一下，似乎效果很不错呢阅读全文

posted @ 2016-03-28 18:32 月儿弯弯0204 阅读(53) | 评论 (0) 编辑

 

flask开发restful api系列(1)

摘要: 这是一篇初步介绍flask的restful api的文章，只是初步文章，后面陆续改进。其中详细介绍了，怎么使用redis在服务器端存储数据方式并且验证。阅读全文