1. Flask介绍
Flask是一个使用 Python 编写的轻量级 Web 应用框架，使用简单操作方便，很适合开发轻量级的API接口，让Android程序员从此不再去问后台要测试接口了。
2. 简单使用
1.安装
pip3 install Falsk
2.使用
from flask import Flask
#初始化
app = Flask(__name__)
#路由
@app.route('/')
def gello_world():
return 'Hello Flask'
if __name__ == '__main__':
#开启service
app.run()
运行这个python文件变启动了服务，默认8080端口
3. 项目实战
实战目标：实现一个注册登陆的接口，JSON格式
1.准备数据库相关
| 字段 | 长度 | 说明 |
|---|---|---|
| id | bigint | 自增长主键ID |
| user | varchar | 用户名 |
| psw | varchar | 用户密码 |
| registerTime | timestamp | 注册时间 |
| phone | varchar | 手机号码 |
2.db.py
数据库使用的是sqlalchemy
# -*- coding:utf-8 -*-
from operator import or_
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, mapper
from sqlalchemy.pool import NullPool
import traceback
'''
解决中文乱码
1.create_engine中添加charset=utf8，encoding='utf-8'
2.表中添加__table_args__ = {
"mysql_charset": "utf8"
}
'''
# 获取实体数据库连接
engine = create_engine('mysql+pymysql://root:1234@127.0.0.1:3306/python?charset=utf8',
encoding='utf-8',
echo=False,
poolclass=NullPool)
metadata = MetaData(engine)
# 反射全库 开始
metadata.reflect(bind=engine)
print('获取所有表名：', metadata.tables.keys())
# 反射全库 结束
Base = automap_base()
Base.prepare(engine, reflect=True)
print('获取所有对象名:', Base.classes.keys())
db_session = sessionmaker(bind=engine)
def login(user_name, user_psw):
'''
登陆
:param user_name:
:param user_psw:
:return:
'''
session = db_session()
db_user = Base.classes.user
try:
user = session.query(db_user).filter(
db_user.user == user_name,
db_user.psw == user_psw).one()
return user
except BaseException:
message = traceback.format_exc()
if not message.__contains__('No row was found for one()'):
#无此用户信息
print(message)
session.close()
return None
def register(user_name, user_psw, user_phone):
'''
注册
:param user_name:
:param user_psw:
:param user_phone:
:return:
'''
session = db_session()
db_user = Base.classes.user
count = session.query(db_user).filter(
or_(db_user.phone == user_phone,
db_user.user == user_name)
).count()
if int(count) > 0:
return False
user = db_user(user=user_name, psw=user_psw, phone=user_phone)
session.add(user)
session.commit()
session.close()
return True
3.servier.py
# 简单的flask接口服务
from flask import Flask, request, jsonify
import sys
import threading
import time
import db
from util import is_phone_num
server = Flask(__name__)
@server.route('/test/login', methods=['GET', 'POST'])
def login():
print(threading.current_thread().name)
username = request.values.get("username")
psw = request.values.get('psw')
if not username or not psw:
return jsonify({'code': 400, 'msg': '参数缺省'})
user = db2.login(username, psw)
if user:
res = {'code': 200, 'msg': '登录成功', 'user': {
'username': user.user,
'phone': user.phone,
'registerTime': str(user.registerTime)
}}
else:
res = {'code': 400, 'msg': '账号或密码错误'}
return jsonify(res)
@server.route('/test/register', methods=['GET', 'POST'])
def register():
username = request.values.get("username")
psw = request.values.get('psw')
phone = request.values.get('phone')
if not username or not psw or not phone:
return jsonify({'code': 400, 'msg': '参数缺省'})
if not is_phone_num(phone):
return jsonify({'code': 400, 'msg': '手机号非法'})
success = db2.register(username, psw, phone)
return jsonify(
{'code': 200 if success else 400, 'msg': '尊敬的{},恭喜您注册成功'.format(username) if success else '用户信息已被注册'})
if __name__ == '__main__':
server.run(
port=8889, #自定义端口
debug=True,#是否开启debug模式
host='192.168.0.105',#指定host
threaded=True#是否开启多线程
)
4.接口
◣
Path:
http://192.168.0.105:8889/test/register?username=TangRen&psw=m123456&phone=18600000000
{
"code": 200,
"msg": "尊敬的TangRen,恭喜您注册成功"
}
◣
Path:
http://192.168.0.105:8889/test/login?username=TangRen&psw=m123456
{
"code": 200,
"msg": "登录成功",
"user": {
"phone": "18600000000",
"registerTime": "2019-06-24 22:11:46",
"username": "TangRen"
}
}
4. 总结
Python大法好
