# _*_ coding: utf-8 _*_
"""
python_restful_api.py by xianhu
"""
import sqlalchemy
import sqlalchemy.orm
import sqlalchemy.ext.declarative
from flask import Flask, g
from flask_restful import reqparse, Api, Resource
from flask_httpauth import HTTPTokenAuth
# Flask相关å˜é‡å£°æ˜Ž
app = Flask(__name__)
api = Api(app)
# 认è¯ç›¸å…³
auth = HTTPTokenAuth(scheme="token")
TOKENS = {
"fejiasdfhu",
"fejiuufjeh"
}
@auth.verify_token
def verify_token(token):
if token in TOKENS:
g.current_user = token
return True
return False
# æ•°æ®åº“相关å˜é‡å£°æ˜Ž
engine = sqlalchemy.create_engine("mysql+pymysql://username:password@ip/db_name", encoding="utf8", echo=False)
BaseModel = sqlalchemy.ext.declarative.declarative_base()
# 构建数æ®æ¨¡åž‹User
class User(BaseModel):
__tablename__ = "Users"
__table_args__ = {
"mysql_engine": "InnoDB",
"mysql_charset": "utf8",
}
# 表结构,具体更多的数æ®ç±»åž‹è‡ªè¡Œç™¾åº¦
id = sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True)
name = sqlalchemy.Column("name", sqlalchemy.String(50), nullable=False)
age = sqlalchemy.Column("age", sqlalchemy.Integer, nullable=False)
# 构建数æ®æ¨¡åž‹çš„jsonæ ¼å¼
def get_json(user):
return {"id": user.id, "name": user.name, "age": user.age}
# 利用Session对象连接数æ®åº“
DBSessinon = sqlalchemy.orm.sessionmaker(bind=engine)
session = DBSessinon()
BaseModel.metadata.drop_all(engine)
BaseModel.metadata.create_all(engine)
# RESTfulAPIçš„å‚æ•°è§£æž -- put / postå‚数解æž
parser_put = reqparse.RequestParser()
parser_put.add_argument("name", type=str, required=True, help="need name data")
parser_put.add_argument("age", type=int, required=True, help="need age data")
# RESTfulAPIçš„å‚æ•°è§£æž -- getå‚数解æž
parser_get = reqparse.RequestParser()
parser_get.add_argument("limit", type=int, required=False)
parser_get.add_argument("offset", type=int, required=False)
parser_get.add_argument("sortby", type=str, required=False)
# æ“作（put / get / delete）å•ä¸€èµ„æº
class Todo(Resource):
# æ·»åŠ è®¤è¯
decorators = [auth.login_required]
def put(self, user_id):
"""
更新用户数æ®: curl http://127.0.0.1:5000/users/1 -X PUT -d "name=Allen&age=20" -H "Authorization: token fejiasdfhu"
"""
args = parser_put.parse_args()
user_ids_set = set([user.id for user in session.query(User.id)])
print(user_ids_set)
# 用户ä¸å˜åœ¨ï¼Œè¿”回404
if user_id not in user_ids_set:
return None, 404
# 更新用户数æ®
user = session.query(User).filter(User.id == user_id)[0]
user.name = args["name"]
user.age = args["age"]
session.merge(user)
session.commit()
# æ›´æ–°æˆåŠŸï¼Œè¿”回201
return get_json(user), 201
def get(self, user_id):
"""
获å–用户数æ®: curl http://127.0.0.1:5000/users/1 -X GET -H "Authorization: token fejiasdfhu"
"""
users = session.query(User).filter(User.id == user_id)
# 用户ä¸å˜åœ¨ï¼Œè¿”回404
if users.count() == 0:
return None, 404
# 返回用户数æ®
return get_json(users[0]), 200
def delete(self, user_id):
"""
åˆ é™¤ç”¨æˆ·æ•°æ®: curl http://127.0.0.1:5000/users/1 -X DELETE -H "Authorization: token fejiasdfhu"
"""
session.query(User).filter(User.id == user_id).delete()
return None, 204
# æ“作（post / get）资æºåˆ—表
class TodoList(Resource):
# æ·»åŠ è®¤è¯
decorators = [auth.login_required]
def get(self):
"""
获å–全部用户数æ®: curl http://127.0.0.1:5000/users -X GET -d "limit=2&offset=0&sortby=name" -H "Authorization: token fejiasdfhu"
"""
args = parser_get.parse_args()
users = session.query(User)
# æ ¹æ®æ¡ä»¶æŸ¥è¯¢
if "sortby" in args:
users = users.order_by(User.name if args["sortby"] == "name" else User.age)
if "offset" in args:
users = users.offset(args["offset"])
if "limit" in args:
users = users.limit(args["limit"])
# 返回结果
return [get_json(user) for user in users], 200
def post(self):
"""
æ·»åŠ ä¸€ä¸ªæ–°ç”¨æˆ·: curl http://127.0.0.1:5000/users -X POST -d "name=Brown&age=20" -H "Authorization: token fejiasdfhu"
"""
args = parser_put.parse_args()
# 构建新用户
user = User(name=args["name"], age=args["age"])
session.add(user)
session.commit()
# 资æºæ·»åŠ æˆåŠŸï¼Œè¿”回201
return get_json(user), 201
# 设置路由
api.add_resource(TodoList, "/users")
api.add_resource(Todo, "/users/