前面通过六部分学习笔记，已经初步从数据模型，数据库更新，序列化，反序列化等搭建了一个基本框架，接下来就是真正的实战了。通过flask_restful来实现一些基本数据库读写操作。
首先回顾一下之前的代码
入口
app.py
此部分代码定义了框架入口主程序。指定了数据库连接方式，并使用flask_restful添加了相关的api。
以下代码中import了api.ResourceAccount模块中的ResourceAccount。
使用flask_restful的Api与flask 对象进行关联绑定，使用Api的add_resource方法绑定URL与Resource类。
from flask import Flask
from model import db
from flask_restful import Api
from api.ResourceAccount import ResourceAccount
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db' # 使用SQLite数据库，可以根据需要更改
db.init_app(app)
api = Api(app)
api.add_resource(ResourceAccount, '/api/account/<string:account_id>')
if __name__ == '__main__':
app.run(debug=True)
# with app.app_context():
# db.create_all()model模块
model.py
定义了Account Profile Project Host几个类
Account 与 Profile 为一对一关系
Account 与 Project 为多对多关系 通过中间表account_project进行关联
Project 与Host 为一对多关系
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Base(db.Model):
"""基类"""
# 作为父类被继承，不会被创建成表（抽象）
__abstract__ = True
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
def save(self):
db.session.add(self)
db.session.commit()
def delete(self):
db.session.delete(self)
db.session.commit()
def update(self):
db.session.commit()
class Account(Base):
__tablename__ = "accounts"
account_name = db.Column(db.String(50), unique=True, nullable=False)
account_email = db.Column(db.String(120), unique=True, nullable=False)
# 与Profile建立一对一关系
profile = db.relationship('Profile', backref='account', uselist=False)
def __init__(self, account_name, account_email):
self.account_name = account_name
self.account_email = account_email
def __repr__(self):
return '<Account %r %r>' % (self.account_name, self.account_email)
class Profile(Base):
__tablename__ = "profiles"
fullname = db.Column(db.String(100))
gender = db.Column(db.String(10))
age = db.Column(db.Integer)
email = db.Column(db.String(100))
height = db.Column(db.Integer)
# 外键
account_id = db.Column(db.Integer, db.ForeignKey('accounts.id'), unique=True)
def __init__(self, fullname, email, gender, age, height):
self.fullname = fullname
self.email = email
self.gender = gender
self.age = age
self.height = height
def __repr__(self):
return '<Profile %r %r %r %r %r>' % (self.fullname, self.email, self.gender, self.age, self.height)
class Project(Base):
__tablename__ = "projects"
project_name = db.Column(db.String(50), unique=True, nullable=False)
project_webhook = db.Column(db.String(150))
# 与Host建立一对多关系
hosts = db.relationship('Host', backref='project', lazy=True)
# 与Accounts通过中间表建立多对多关系
accounts = db.relationship('Account', secondary='account_project', backref=db.backref('projects', lazy=True))
def __init__(self, project_name, project_webhook):
self.project_name = project_name
self.project_webhook = project_webhook
def __repr__(self):
return '<Project %r %r %r %r>' % (self.project_name, self.project_webhook, self.hosts, self.accounts )
class Host(Base):
__tablename__ = "hosts"
hostname = db.Column(db.String(50))
ip = db.Column(db.String(15))
project_id = db.Column(db.Integer, db.ForeignKey('projects.id'))
def __init__(self, hostname, ip):
self.hostname = hostname
self.ip = ip
def __repr__(self):
return '<Host %r %r>' % (self.hostname, self.ip)
# 中间表
account_project = db.Table(
'account_project',
db.Column('account_id', db.Integer, db.ForeignKey('accounts.id')),
db.Column('project_id', db.Integer, db.ForeignKey('projects.id'))
)schema模块
schema.py
schema与model内容相对应
from marshmallow import Schema, fields, validate, validates_schema, ValidationError, validates
class BaseSchema(Schema):
id = fields.Integer(dump_only=True)
class AccountSchema(BaseSchema):
account_name = fields.String(required=True)
account_email = fields.String(required=True)
profile = fields.Nested('ProfileSchema', exclude=('account',))
projects = fields.Nested('ProjectSchema', many=True, exclude=('accounts',))
class ProfileSchema(BaseSchema):
fullname = fields.String()
gender = fields.String()
account = fields.Nested('AccountSchema', exclude=('profile',))
age = fields.Int(validate=validate.Range(min=18, max=65))
email = fields.Email()
height = fields.Int()
@validates_schema
def validate_profile_gender(self, data, **kwargs):
if data['gender'] not in ['M', 'F', 'f', 'm']:
raise ValidationError(field_name='gender', message='gender must be M or F')
else:
data['gender'] = data['gender'].upper()
return data
@validates('height')
def validate_age(self, data, **kwargs):
if data <= 150:
raise ValidationError('height must be taller than 150')
else:
return data
class ProjectSchema(BaseSchema):
project_name = fields.String(required=True)
project_webhook = fields.String()
hosts = fields.Nested('HostSchema', many=True, exclude=('project',))
accounts = fields.Nested('AccountSchema', many=True, exclude=('projects',))
class HostSchema(BaseSchema):
hostname = fields.String()
ip = fields.String()
project = fields.Nested('ProjectSchema', exclude=('hosts',))管理模块
manager.py
迁移命令管理与app,建立关系
Resource类
api/ResourceAccount.py
在ResourceAccount中定义了四个常用的方法 get post put patch，具体详解见下文
from flask import request
from flask_restful import Resource
from model import *
from schema import *
class ResourceAccount(Resource):
def get(self, account_id):
"""
GET方法 根据account_id获取Account信息
"""
if account_id:
a = Account.query.filter_by(id=account_id).first()
print(a)
if a:
a_s = AccountSchema()
print(a_s.dump(a))
return {"success": True, "data": a_s.dump(a)}
else:
return {"success": False, "message": "Account not found"}, 404
else:
a = Account.query.all()
r = []
for i in a:
a_s = AccountSchema()
r.append(a_s.dump(i))
if len(r) == 0:
return {"success": False, "message": "No accounts found"}, 404
else:
return {"success": True, "data": r}
def post(self):
"""
POST方法 创建Account
"""
json_data = request.get_json(force=True)
if not json_data:
return {"success": False, "message": "No input data provided"}, 400
else:
a_s = AccountSchema()
data, errors = a_s.load(json_data)
if errors:
return {"success": False, "message": "Invalid input data", "errors": errors}, 422
else:
a = Account(**data)
db.session.add(a)
db.session.commit()
return {"success": True, "data": a_s.dump(a)}, 201
def patch(self, account_id):
"""
PATCH方法 更新Account
部分更新
"""
json_data = request.get_json(force=True)
if not json_data:
return {"success": False, "message": "No input data provided"}, 400
else:
a_s = AccountSchema()
# put与patch最大的区别是全量更新与部分更新。
# 因为传递过来的数据可能是不完整的，所以在使用load的时候要添加partial参数。
# 如果 partial 参数设置为 True，Marshmallow 将允许部分加载的 JSON 数据。
# 这意味着即使 JSON 数据不完整或缺少某些字段，Marshmallow 仍然会尝试加载可用的数据部分。
data, errors = a_s.load(json_data, partial=True)
if errors:
return {"success": False, "message": "Invalid input data", "errors": errors}, 422
else:
a = Account.query.filter_by(id=account_id).first()
if a:
for k, v in data.items():
setattr(a, k, v)
db.session.commit()
return {"success": True, "data": a_s.dump(a)}, 200
else:
return {"success": False, "message": "Account not found"}, 404
def put(self, account_id):
"""
PUT方法 更新Account
全量更新
"""
a = Account.query.filter_by(id=account_id).first()
if a:
json_data = request.get_json(force=True)
if not json_data:
return {"success": False, "message": "No input data provided"}, 400
else:
a_s = AccountSchema()
data, errors = a_s.load(json_data)
if errors:
return {"success": False, "message": "Invalid input data", "errors": errors}, 422
else:
a.update(**data)
db.session.commit()
return {"success": True, "data": a_s.dump(a)}, 200
else:
return {"success": False, "message": "Account not found"}, 404
def delete(self, account_id):
"""
DELETE方法 删除Account
Delete account
"""
a = Account.query.filter_by(id=account_id).first()
if a:
db.session.delete(a)
db.session.commit()
return {"success": True, "message": "Account deleted"}, 200
else:
return {"success": False, "message": "Account not found"}, 404试运行
运行app.py
在浏览器中访问127.0.0.1:5000/api/account/1 可会获得以下结果（数据在前面的文档中已经有过验证）
