sqlalchemy2 学习进度select完成

main
roger_home_pc 1 year ago
parent 23cf8bb940
commit 9adfd58551
  1. 52
      sqlalchemy_20/db.py

@ -1,4 +1,4 @@
from sqlalchemy import create_engine, text, insert, update, select, delete, engine_from_config, or_, and_, func
from sqlalchemy import create_engine, text, insert, update, select, delete, engine_from_config, or_, and_, func, any_
from sqlalchemy.orm import Session, sessionmaker
import configparser
import sys
@ -246,18 +246,54 @@ insert_data2 = {'name': 'sql2_test2', 'password': 'pwd2'}
# query_result = db_session.execute(sql_stmt).scalar()
# print(query_result)
# sqlalchemy1.4:
# 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__ 后可以打印
# try:
# sql_stmt = select(User).where(User.id > 2)
# query_result = db_session.execute(sql_stmt).one()
# except Exception as e:
# print(e)
# sqlalchemy1.4: session.query(User).filter(User.id == 1).one_or_none()
# try:
# # sql_stmt = select(User).where(User.id >= 1) # 多结果
# # sql_stmt = select(User).where(User.id == 1) # 一个结果
# sql_stmt = select(User).where(User.id == 100) # 无结果
# query_result = db_session.execute(sql_stmt).one_or_none()
# print(query_result)
# except Exception as e:
# print(e)
# sqlalchemy1.4: session.query(User).filter(User.id == 1).scalar()
# sql_stmt = select(User).where(User.id >= 1)
# query_result = db_session.execute(sql_stmt).scalar()
# print(query_result.name)
# sqlalchemy1.4: session.query(User).filter(User.id > 1).group_by(User.password)
# sql_stnt = select(User.password).where(User.id > 1).group_by(User.password)
# query_result = db_session.execute(sql_stnt).scalars()
# for row in query_result:
# print(row)
# sqlalchemy1.4: session.query(func.max(User.id)).all()
# sql_stmt = select(func.max(User.id))
# query_result = db_session.execute(sql_stmt).scalars()
# for row in query_result:
# print(row)
# func
# func.now()
# func.current_timestamp()
# func.current_date()
# func.current_time()
# func.min()
# func.localtime()
# func.localtimestamp()
# func.random()
# func.sum()
# select all scalars
# result = db_session.execute(select(User)).scalars().all() # [<1 name:rogersun>, <2 name:honeyhoney>, <3 name:haha>]

Loading…
Cancel
Save