from flask import Flask from sqlalchemy import not_,or_,and_,desc from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)
class RoleDB(db.Model): __tablename__ = "role" id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(16), unique=True)
users = db.relationship('UserDB', backref="role")
def __repr__(self): return "Role: %s | %s" % (self.id, self.name)
class UserDB(db.Model): __tablename__ = "user" id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(16), unique=True) email = db.Column(db.String(32), unique=True) password = db.Column(db.String(16))
role_id = db.Column(db.Integer, db.ForeignKey("role.id"))
def __repr__(self): return "User: %s | %s | %s | %s" % (self.id, self.name, self.password, self.role_id)
@app.route('/') def index(): print("查询所有用户数据: ", UserDB.query.all()) print("查询所有规则数据: ", RoleDB.query.all()) print("查询有多少个用户: ", UserDB.query.count()) print("查询第一个用户: ", UserDB.query.first())
print("查询第一条: ", UserDB.query.get(1)) print("查询过滤器(类名+属性名): ", UserDB.query.filter(UserDB.id == 4).first()) print("查询过滤器(属性名): ", UserDB.query.filter_by(id=4).first())
print(UserDB.query.filter(UserDB.name.endswith("g")).all()) print(UserDB.query.filter(UserDB.name.contains("g")).all())
print(UserDB.query.filter(not_(UserDB.name == "wang")).all()) print(UserDB.query.filter(UserDB.name != "wang").all())
print(UserDB.query.filter(and_(UserDB.name.startswith("li"),UserDB.email.startswith("li"))).all()) print(UserDB.query.filter(UserDB.name.startswith("li"), UserDB.email.startswith("li")).all())
print(UserDB.query.filter(or_(UserDB.password=='123456', UserDB.email.endswith('lyshark.com'))).all())
print(UserDB.query.filter(UserDB.id.in_([1, 3, 5, 7, 9])).all())
print(UserDB.query.filter_by(name="liu").first().role.name)
print("升序: ", UserDB.query.order_by("email").all()) print("降序: ", UserDB.query.order_by(desc("email")).all())
pages = UserDB.query.paginate(2,3,False) print("查询结果: {} 总页数: {} 当前页数: {}".format(pages.items,pages.pages,pages.page))
ref = db.session.query(UserDB).filter(UserDB.name == "wang").all() print(ref) return "success"
if __name__ == "__main__": db.drop_all() db.create_all()
role_admin = RoleDB(name="admin") db.session.add(role_admin) role_lyshark = RoleDB(name="lyshark") db.session.add(role_lyshark) db.session.commit()
try: ua = UserDB(name='wang', email='wang@163.com', password='123456', role_id=role_admin.id) ub = UserDB(name='zhang', email='zhang@189.com', password='201512', role_id=role_lyshark.id) uc = UserDB(name='chen', email='chen@126.com', password='987654', role_id=role_lyshark.id) ud = UserDB(name='zhou', email='zhou@163.com', password='456789', role_id=role_admin.id) ue = UserDB(name='tang', email='tang@lyshark.com', password='158104', role_id=role_lyshark.id) uf = UserDB(name='wu', email='wu@gmail.com', password='5623514', role_id=role_lyshark.id) ug = UserDB(name='qian', email='qian@gmail.com', password='1543567', role_id=role_admin.id) uh = UserDB(name='liu', email='liu@lyshark.com', password='867322', role_id=role_admin.id) ui = UserDB(name='li', email='li@163.com', password='4526342', role_id=role_lyshark.id) uj = UserDB(name='sun', email='sun@163.com', password='235523', role_id=role_lyshark.id) db.session.add_all([ua, ub, uc, ud, ue, uf, ug, uh, ui, uj]) db.session.commit()
except Exception as e: db.session.rollback() raise e
app.run(debug=True)
|