完成sqlalchemy 1.4的增删改查学习

sqlalchemy_1.4/db_schema.py

@@ -1,6 +1,7 @@
 import sqlalchemy
-from sqlalchemy.orm import declarative_base, sessionmaker, Session  # declarative 声明式
+from sqlalchemy.orm import declarative_base, sessionmaker, Session, relationship, backref  # declarative 声明式
-from sqlalchemy import create_engine, Column, Integer, String
+from sqlalchemy import create_engine, Column, Integer, String, DATE, Enum, desc, or_, and_, update, case, union, \
+    union_all
 
 # 创建数据引擎
 HOST = "home.rogersun.cn"
@@ -9,13 +10,14 @@ PWD = "Sxzgx1209"
 PORT = "3306"
 DB = "orm_sqlalchemy"
 
-URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}"
+URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}?charset=utf8"
 engine = create_engine(URI, echo=True)
 
 # 创建session
-session = sessionmaker(engine)
+SessionClass = sessionmaker(bind=engine)
+session = SessionClass()
 
-# with session
+# # with Session
 # with Session(engine) as db_session:
 #     session.begin()
 #     try:
@@ -31,5 +33,130 @@ session = sessionmaker(engine)
 Base = declarative_base()
 
 
+# 创建表结构
 class User(Base):
     __tablename__ = 'user'
+    id = Column(Integer, primary_key=True)
+    name = Column(String(32))
+    password = Column(String(64))
+
+    def __repr__(self):
+        return "<%s name:%s>" % (self.id, self.name)
+
+
+class Student(Base):
+    __tablename__ = 'student'
+    id = Column(Integer, primary_key=True)
+    name = Column(String(32), nullable=False)
+    gender = Column(Enum('m', 'f'), nullable=False)
+    date = Column(DATE, nullable=False)
+
+    def __repr__(self):
+        return f"<{self.id}-{self.name}-{self.gender}-{self.date}>"
+
+
+# 生成实体表
+# Base.metadata.create_all(engine)
+
+# insert
+# 创建数据
+user_data_1 = User(name="roger", password="123321")
+user_data_2 = User(name="honey", password='123456')
+#
+# # 插入数据
+# session.add(user_data_1)
+# session.add(user_data_2)
+#
+# # 提交数据
+# session.commit()
+# insert完整示例
+# try:
+#     session.add(user_data_1)
+# except Exception as e:
+#     print(e)
+#     session.rollback()
+# else:
+#     session.commit()
+
+
+# select
+# result = session.query(User).all()  # all
+# result = session.query(User).filter_by(name='roger').all()  # filter_by
+# result = session.query(User).filter(User.id == 2).all()  # filter
+# result = session.query(User).filter(User.id >= 2).first()  # first
+# result = session.query(User).filter(User.id >= 2).order_by(User.id.desc()).all()  # order_by desc
+# result = session.query(User).filter(User.id >= 2).filter(User.name.like('%one%')).all()  # like
+# result = session.query(User).filter(User.id.in_([1, 2, 3])).all()  # in
+# # in startswith
+# result = session.query(User).filter(
+#     User.id.in_(session.query(User.id).filter(User.name.startswith('he')))
+# ).all()
+# result = session.query(User).filter(User.id.notin_([1, 2])).all()  # notin
+# result = session.query(User).filter(User.password is None).all()  # is None
+# result = session.query(User).filter(User.password.is_(None)).all()  # is_(None)
+# result = session.query(User).filter(User.password is not None).all()  # is not None
+# result = session.query(User).filter(User.password.isnot(None)).all()  # isnot(None)
+# result = session.query(User).filter(or_(User.id > 2, User.name.ilike('Rog%'))).all()  # or_
+# result = session.query(User).filter(and_(User.id >= 2, User.name.endswith('ei'))).all()  # and_ endswith
+# result = session.query(User.name.label('user_name'), User.password).filter(User.id >= 2).all()  # label  print(result[0].user_name)
+# result = session.query(User).filter(User.id > 1).count()  # count
+# try:
+#     result = session.query(User).filter(User.id > 1).one()  # one 当仅查询到一个结果时返回，否则抛出异常
+# except Exception as e:
+#     print(e)
+# result = session.query(User).filter(User.id == 1).one_or_none()  # one_or_none 当查询结果仅一条时显示返回结果，当没有结果时返回None，否则抛出异常
+# result = session.query(User).filter(User.id == 1).scalar()  # 调用one方法，并在成功时返回行的第一列
+result = session.query(User).filter(User.id > 1).group_by(User.password)
+# print(result[0].name, result[0].password)
+print(result)  # 定义 __repr__ 后可以打印
+
+# join
+# filter形式
+# result = session.query(User, Student).filter(User.name == Student.name).all()
+# print(result[0].User.name, result[0].User.password, result[0].Student.name, result[0].Student.gender,
+#       result[0].Student.date)
+#
+# # join形式 无外键关联的情况
+# result = session.query(User).join(Student, User.name == Student.name).all()
+# print(result[0].name, result[0].password)
+
+# join有外键关联的情况（Todo）
+
+
+# union
+# union
+# q1 = session.query(User.name).filter(User.id > 1)
+# q2 = session.query(Student.name).filter(Student.id <= 2)
+# result = q1.union(q2).all()
+# print(result)
+
+# union_all
+# q1 = session.query(User.id, User.name)
+# q2 = session.query(Student.id.label('id'), Student.name.label('name'))
+# result = q1.union_all(q2).filter(User.id <= 2)
+# print(dict(result))
+# for r in result:
+#     print(r)
+
+
+# # update
+# 一般形式
+# data = session.query(User).filter(User.id == 1).first()
+# print(data)
+# data.name = 'rogersun'
+# session.commit()
+# 连写形势
+# data = session.query(User).filter(User.id == 2).update({'name': 'honeyhoney'})
+# session.commit()
+# 批量更新
+# data = session.query(User).filter(User.password.isnot(None)).update({User.password: '123321'})
+# print(data)
+# session.commit()
+
+# delete
+result = session.query(User).filter(User.name == 'delete').delete()
+print(result)
+session.commit()
+
+# 关闭session
+session.close()