/model/user_model.py

from flask import Blueprint

from model.base_model import Session
from model.user_model import User

user_blueprint = Blueprint('user', __name__)


@user_blueprint.route('/user/add')
def user_add():
    user = User(name='a', age=1)

    session = Session()
    session.add(user)
    session.commit()
    return '用户-add'


@user_blueprint.route('/user/del')
def user_del():
    session = Session()
    user = session.query(User).get(1)

    session.delete(user)
    session.commit()
    return '用户-del'


@user_blueprint.route('/user/update')
def user_update():
    session = Session()
    user = session.query(User).get(2)
    user.age = 22
    session.commit()
    return '用户-update'


@user_blueprint.route('/user/list')
def user_list():
    session = Session()
    res = session.query(User).all()

    rows = []
    for item in res:
        row = {}
        for k, v in item.__dict__.items():
            if k != '_sa_instance_state':
                row[k] = v
        rows.append(row)
    return rows


@user_blueprint.route('/user/info')
def user_info():
    session = Session()
    res = session.query(User).get(2)

    user_dict = res.__dict__
    del user_dict['_sa_instance_state']

    return user_dict

/model/base_model.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

# 创建数据库连接
db_host = 'xxx.xxx.xxx.xxx'
db_user = 'xys-test'
db_pass = 'xys-test'
db_database = 'xys-test'

engine = create_engine('mysql+pymysql://%s:%s@%s:3306/%s' % (db_user, db_pass, db_host, db_database))

# 创建 scoped session
Session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

# 创建基础模型
Base = declarative_base()