Sqlalchemy 单表操作基本用法

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2019/4/3 14:48
# @Author  : Ropon
# @File    : sqlalchemy.py

# 导入官宣基础模型
from sqlalchemy.ext.declarative import declarative_base
# 实例化官宣基础模型
Base = declarative_base()
class User(Base):
    # 配置表名
    __tablename__ = "user"
    from sqlalchemy import Column, Integer, String
    # id = Column()
    # int = Integer
    # 配置列 整型 主键 自增
    nid = Column(Integer, primary_key=True, autoincrement=True)
    # 同理配置列 字符串 创建索引
    name = Column(String(32), index=True)


from sqlalchemy import create_engine
# 创建数据库引擎
engine = create_engine("mysql+pymysql://sqlormtest:xxxxxx@x.x.x.x:3306/sqlormtest?charset=utf8")
# mysql+pymysql：指定链接那种类型数据库
# sqlormtest：用户名
# xxxxxx：sqlormtest用户的密码
# x.x.x.x:3306 数据库的IP及端口号
# sqlormtest 数据库名
# charset=utf8 数据库字符集


Base.metadata.create_all(engine)

###################################################################
from sqlormtest import User
# 创建一个操作会话
from sqlalchemy.orm import sessionmaker
# 导入之前创建数据库引擎
from sqlormtest import engine

Session = sessionmaker(engine)
db_session = Session()
#############################################
# 添加数据
user1 = User(name="Ropon")
# db_session会话中添加一条 User ORM模型创建的数据
# db_session.add(user1)
# 将db_session 会话所有指令一次性提交
# db_session.commit()
# 关闭会话
# db_session.close()

# user_list = [
#     User(name="Ro"),
#     User(name="Lp"),
#     User(name="Pg")
# ]
# db_session.add_all(user_list)
# db_session.commit()
# db_session.close()
#############################################
# 查询数据
# user_all_list = db_session.query(User).all()
# user_all_list = db_session.query(User).filter(User.nid >= 3).all()
# print(user_all_list)
# for obj in user_all_list:
#     print(obj.nid, obj.name)

# user = db_session.query(User).filter(User.nid >= 3).first()
# print(user.nid, user.name)

# wulong1 = db_session.query(User).filter(User.nid >= 3)
# print(wulong1)
#
# wulong2 = db_session.query(User)
# print(wulong2)

#############################################
# 修改数据

# res = db_session.query(User).filter(User.nid == 3).update({"name": "LuoPeng"})
# print(res)
# db_session.commit()
# db_session.close()

#############################################
# 删除数据

# res = db_session.query(User).filter(User.nid==2).delete()
# print(res)
# db_session.commit()
# db_session.close()

#############################################
# 高级操作  查询
from sqlalchemy.sql import and_, or_
# res = db_session.query(User).filter(and_(User.nid > 2, User.name == "pengge")).first()
# print(res.name)

# res = db_session.query(User).filter(or_(User.nid < 3, User.name == "pengge")).all()
# print(res)

# 指定列做个别名 name as username
# res = db_session.query(User.name.label("username"), User.nid).first()
# print(res.nid, res.username)

# res = db_session.query(User).filter(User.name == "ropon").all()
# print(res)

# 原生SQL筛选条件
# res = db_session.query(User).filter_by(name="luopeng").all()
# res2 = db_session.query(User).filter_by(name="luopeng").first()
# print(res)
# print(res2.name)

# 字符串匹配筛选条件 order_by 进行排序
from sqlalchemy.sql import text

# res = db_session.query(User).filter(text("nid<:value and name=:name")).params(value=3, name="luopeng").order_by(
#     User.nid).first()
# print(res.nid)

# 原生SQL语句查询?
# res = db_session.query(User).filter(text("select * from User id>:value"))
# print(res)

# 筛选查询列
# res = db_session.query(User.name).first()
# print(res.name)

# 排序 默认升序 加desc()方法降序
# user_list = db_session.query(User.nid).order_by(User.nid.desc()).all()
# print(user_list)

# res = db_session.query(User).filter_by(name="ropon").all()
# print(res)
# 且
# res = db_session.query(User).filter(User.nid > 1, User.name == "pengge").all()
# between(1, 3) 大于1小于3
# res2 = db_session.query(User).filter(User.nid.between(1, 3), User.name == "pengge").all()
# print(res)

# 只查询nid等于1,3,4
# res = db_session.query(User).filter(User.nid.in_([1, 3, 4])).all()
# 查询nid不等于1,2,4
# res2 = db_session.query(User).filter(~User.nid.in_([1, 2, 4])).all()
# print(res2[0].nid)

# 子查询
# res = db_session.query(User).filter(User.nid.in_(db_session.query(User.nid).filter_by(name="pg"))).all()
# print(res[0].name)

# 且 或
# from sqlalchemy import and_, or_
# res = db_session.query(User).filter(and_(User.nid > 3, User.name == "ropon")).all()
# res2 = db_session.query(User).filter(or_(User.nid > 2, User.name == "ropon")).all()
# print(res2)

# 通配符 % 取反 ~
# 限制 [1:2]
# res = db_session.query(User).filter(User.name.like("ropon%"))[1:2]
# print(res)

from sqlalchemy.sql import func
# res = db_session.query(User.name).group_by(User.nid).all()
# res = db_session.query(func.max(User.nid)).group_by(User.name).all()
# print(res)


#############################################
# 高级操作  修改

# res = db_session.query(User).filter(User.nid>2).update({"name": "Pgg"})
# res1 = db_session.query(User.nid, User.name).all()
# print(res1)

# db_session.query(User).filter(User.nid>3).update({User.name: User.name + "GoodBoy"}, synchronize_session=False)
db_session.query(User).filter(User.nid<2).update({User.name: User.name + 123}, synchronize_session="evaluate")

db_session.commit()
db_session.close()