orm_sqlalchemy/sqlalchemy_14/db_schema.py

import sqlalchemy
from sqlalchemy.orm import declarative_base, sessionmaker, Session, relationship, backref  # declarative 声明式
from sqlalchemy import create_engine, Column, Integer, String, DATE, Enum, func, desc, or_, and_, update, case, union, \
    union_all

# 创建数据引擎
HOST = "home.rogersun.cn"
USER = "root"
PWD = "Sxzgx1209"
PORT = "3306"
DB = "orm_sqlalchemy"

URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}?charset=utf8"
engine = create_engine(URI, echo=True)

# 创建session
SessionClass = sessionmaker(bind=engine)
session = SessionClass()

# # with Session
# with Session(engine) as db_session:
#     session.begin()
#     try:
#         db_session.add("some_object")
#         db_session.add("some_other_object")
#     except:
#         db_session.rollback()
#         raise
#     else:
#         db_session.commit()

# 生成orm基类
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)
# result = session.query(func.max(User.id)).all()  # func.max
# 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()