flask mysql sqlalchemy教程

import logging
import ast
import simplejson as json
import time
from flask_sqlalchemy import SQLAlchemy
from flask import Flask

from commons.entity import rds_mongo

app = Flask(__name__)

app.config[‘SQLALCHEMY_DATABASE_URI‘] = ‘mysql://root:1234@localhost:3306/flask_demo‘

app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS‘] = True
db = SQLAlchemy(app)

logger = logging.getLogger("platform")

class UserInfo(DynamicDocument):
    # Columns

    __tablename__ = ‘userinfo‘

    uid = Column(String(64), primary_key=True)
    nick_name = Column(String(64))
    sex = Column(String(64))
    email = Column(String(64))
    device_id = Column(String(100))
    imei = Column(String(100))
    coin_balance = Column(Integer())  # 金币余额
    real_name = Column(String(100))  # 真名
    id_card = Column(String(100))  # 身份证号
    create_time = Column(String(100))  # 创建时间
    last_login_time = Column(String(100), nullable=True)  # 创建时间
    cash = Column(Float())  # # 现金账户
    other_column = Column(Text(), nullable=True)

    def __init__(self, coin_balance=0, lx_login=1, lj_login=1, *args, **kwargs):
        super(UserInfo, self).__init__(*args, **kwargs)
        self.uid = uid_value_decorator(random.randint(0, 9))
        self.coin_balance = coin_balance
        self.lx_login = lx_login  # 连续登陆的天数
        self.lj_login = lj_login  # 累计登陆的天数

# 查询所有
def checkAll():
    user =UserInfo.query.all()
    print(user)


# 修改数据
def update(name, pwd):
    user = User.query.filter(User.name == name).first()
    if user is not None:
        user.thrust = pwd
        db.session.commit()
        print(‘修改 用户 %s ,密码为：%s‘ % (name, pwd))
    else:
        print(‘用户不存在‘)


# 删除数据
def delete(name, pwd):
    user = User.query.filter(User.name == name, User.thrust == pwd).first()
    if user is not None:
        db.session.delete(user)
        db.session.commit()
        print(‘删除 用户 %s ,密码为：%s‘ % (name, pwd))
    else:
        print(‘用户不存在，或密码不正确‘)

from models  import User
#添加数据
user=UserInfo(1,‘xiaoxiao‘,‘ss123‘)
user.save()
#按条件查询
result=User.query.filter(User.id>5).all()
result=User.query.filter(User.username==‘xiaoxiao‘).all() #返回结果为一个列表,列表内元素为User对象,all()为返回查询的所有结果,first()返回查询结果中的第一个
result=User.query.filter(User.username.startswith(‘x‘))  #starstwith以什么开头
#获取查询结果的总数量
count=User.query.filter(User.id>5).count()
#获取查询结果中指定的数据
result=User.query.filter(User.id>5).all()[1:3]  #查询结果以列表返回,所以可以根据列表的切片操作来获取对应数据

#多条件查询
#sqlalchemy内置了多条件查询方法 : and_(), or()_ ,not_()
result=User.query.filter(and_(User.id>5,User.username.startswith(‘x‘))) #查询id大于5并且用户名以x开头的
result=User.query.filter(or_(User.id>5,User.username.startswith(‘x‘))) #查询id大于5或者用户名以x开头的
result=User.query.filter(and_(User.username.startswith(‘x‘))) #查询用户名不是以x开头的


#修改
result=User.query.filter(User.username==‘xiaoxiao‘).all()[0]
result.password=‘000000‘
db.session.commit()

#删除
result=User.query.filter(User.username==‘xiaoxiao‘).all()[0]
db.session.delete(result)
db.session.commit()