From 23cf8bb940d46022a22b8d313cae19cda29e44ed Mon Sep 17 00:00:00 2001 From: roger Date: Sun, 25 Jun 2023 15:33:13 +0800 Subject: [PATCH] =?UTF-8?q?2.0=E5=AD=A6=E4=B9=A0=E8=BF=9B=E5=BA=A6=2080%?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sqlalchemy_20/db.py | 166 +++++++++++++++++++++++++++++++++++----- sqlalchemy_20/db_obj.py | 3 +- 2 files changed, 149 insertions(+), 20 deletions(-) diff --git a/sqlalchemy_20/db.py b/sqlalchemy_20/db.py index 9d9f01d..22a3f97 100644 --- a/sqlalchemy_20/db.py +++ b/sqlalchemy_20/db.py @@ -1,4 +1,4 @@ -from sqlalchemy import create_engine, text, insert, update, select, delete, engine_from_config +from sqlalchemy import create_engine, text, insert, update, select, delete, engine_from_config, or_, and_, func from sqlalchemy.orm import Session, sessionmaker import configparser import sys @@ -13,7 +13,7 @@ PWD = "Sxzgx1209" PORT = "3306" DB = "orm_sqlalchemy" -URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}?charset=utf8" +URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}?charset=utf8mb4" engine = create_engine( URI, @@ -59,22 +59,22 @@ engine = create_engine( # conn.commit() # 获取结果 -stmt = text("SELECT * FROM user") -with engine.connect() as conn: - result = conn.execute(stmt) - # 解包赋值 - # for _id, name, password in result: - # print(f"name: {name}, password: {password}") - # 列指定的名称 - # for row in result: - # name, password = row.name, row.password - # print(f"name: {name}, password: {password}") - # 绑定模型 - # result: [User] = result.fetchall() - # for row in result: - # name = row.name - # password = row.password - # print(f"name: {name}, password: {password}") +# stmt = text("SELECT * FROM user") +# with engine.connect() as conn: +# result = conn.execute(stmt) +# 解包赋值 +# for _id, name, password in result: +# print(f"name: {name}, password: {password}") +# 列指定的名称 +# for row in result: +# name, password = row.name, row.password +# print(f"name: {name}, password: {password}") +# 绑定模型 +# result: [User] = result.fetchall() +# for row in result: +# name = row.name +# password = row.password +# print(f"name: {name}, password: {password}") # 创建Session实例 # with Session(engine) as session: @@ -120,17 +120,145 @@ insert_data2 = {'name': 'sql2_test2', 'password': 'pwd2'} # insert 对象方式 批量 # insert_obj2 = User(name='sql2_test8', password='pwd8') -# insert_obj3 = User(name='sql2_test9', password='pwd9') # with db_session as sess: # # sess.bulk_save_objects([insert_obj2, insert_obj3]) # sess.add_all([insert_obj2, insert_obj3]) # sess.commit() +# add +# with db_session as sess: +# user_obj = User(name='add', password='addpwd123') +# sess.add(user_obj) +# sess.flush() # 刷新数据库表 +# print(user_obj.id) # 获取数据对应id +# sess.commit() + # 查 + # get 通过id # result = db_session.get(User, 4) # print(result) +# sqlalchemy1.4: session.query(User).all() +# query_cursor = db_session.execute(select(User)) +# print(query_cursor.all()) # 数据指针，当all取过数据后one将得到None +# print(query_cursor.first()) +# for row in query_cursor.scalars(): +# print(row) # row.name格式可以取到数据 + +# sqlalchemy1.4: session.query(User).filter_by(name='roger').all() +# sql_stmt = select(User).filter_by(name='roger') +# query_result = db_session.execute(sql_stmt).one() +# print(query_result) +# print(query_result[0].name) + +# sqlalchemy1.4: session.query(User).filter(User.id == 2).all() +# sql_stmt = select(User).where(User.id == 2) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.id >= 2).first() +# sql_stmt = select(User).where(User.id >= 2) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.id >= 2).order_by(User.id.desc()).all() +# sql_stmt = select(User).where(User.id >= 2).order_by(User.id.desc()) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.id >= 2).filter(User.name.like('%one%')).all() +# sql_stmt = select(User).where(User.id >= 2).where(User.name.like('%one%')) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.id.in_([1, 2, 3])).all() +# sql_stmt = select(User).where(User.id.in_([1, 2, 3])) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.id.in_(session.query(User.id).filter(User.name.startswith('he')))).all() +# 连写 +# sql_stmt = select(User).where(User.id.in_( +# db_session.execute(select(User.id).where(User.name.startswith('he'))).scalars())) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") +# 拆分 +# sql_stmt1 = select(User.id).where(User.name.startswith('he')) +# query_result1 = db_session.execute(sql_stmt1).scalars() +# sql_stmt2 = select(User).where(User.id.in_(query_result1)) +# query_result2 = db_session.execute(sql_stmt2).scalars() +# for row in query_result2: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.id.notin_([1, 2])).all() +# sql_stmt = select(User).where(User.id.notin_([1, 2])) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.password is None).all() +# sqlalchemy1.4: session.query(User).filter(User.password.is_(None)).all() +# sql_stmt = select(User).where(User.password.is_(None)) # 此处不支持 is None +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(User.password is not None).all() +# sqlalchemy1.4: session.query(User).filter(User.password.isnot(None)).all() +# sql_stmt = select(User).where(User.password.isnot(None)) # isnot() +# sql_stmt = select(User).where(User.password.is_not(None)) # is_not() +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(or_(User.id > 2, User.name.ilike('Rog%'))).all() +# sql_stmt = select(User).where(or_(User.id > 2, User.name.ilike('ROG%'))) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User).filter(and_(User.id >= 2, User.name.endswith('ei'))) +# sql_stmt = select(User).where(and_(User.id >= 2, User.name.endswith('ei'))) +# query_result = db_session.execute(sql_stmt).scalars() +# for row in query_result: +# print(f"{row.id} - {row.name} - {row.password}") + +# sqlalchemy1.4: session.query(User.name.label('user_name'), User.password).filter(User.id >= 2).all() +# sql_stmt = select(User.name.label('user_name_label'), User.password.label('user_pwd_label')).where(User.id >= 2) +# query_result = db_session.execute(sql_stmt).all() # 返回 [('honeyhoney', '2222'), ('haha', '123321'), ('hehe', '3333')] +# query_result = db_session.execute(sql_stmt).scalars() # 返回 scalars 对象， 仅包含name +# query_result = db_session.execute(sql_stmt) # 返回 {'user_name_label': 'roger', 'user_pwd_label': 'rogerpwd123'} 建议此情况 +# print(query_result) +# for row in query_result: +# print(row) +# for row in query_result.mappings(): +# print(row['user_name_label'], row['user_pwd_label']) + +# sqlalchemy1.4: session.query(User).filter(User.id > 1).count() +# sql_stmt = select(func.count()).select_from(User).where(User.id <= 2) +# query_result = db_session.execute(sql_stmt).scalar() +# print(query_result) + + +# 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__ 后可以打印 + + # select all scalars # result = db_session.execute(select(User)).scalars().all() # [<1 name:rogersun>, <2 name:honeyhoney>, <3 name:haha>] # print(result) diff --git a/sqlalchemy_20/db_obj.py b/sqlalchemy_20/db_obj.py index 42e0359..384368a 100644 --- a/sqlalchemy_20/db_obj.py +++ b/sqlalchemy_20/db_obj.py @@ -27,7 +27,8 @@ class Student(Base): return f"<{self.id}-{self.name}-{self.gender}-{self.date}>" -Base.metadata.create_all(engine) +Base.metadata.create_all(engine) # 创建全部表 +User.__table__.create(engine) # 创建单个表 # 创建表结构 -- 命令式