Select专题 - 准备

# 导入引擎创建函数 from sqlalchemy import create_engine # 导入语句处理函数 from sqlalchemy.sql.expression import text # 导入SQL构造相关函数 ''' from sqlalchemy.sql.expression import ( select, insert, delete, udpate, table, column, func, asc, desc, between, distinct, cast, case, literal_column, bindparam, and_, or_, not_, all_, any_ ) ''' # 不单独加载每个函数，只引入模块，后续按需调用 from sqlalchemy.sql import expression as sse # 导入元数据、表类 from sqlalchemy import MetaData, Table # 导入数据类型 from sqlalchemy import Integer, Float, Text, String, TIMESTAMP, DateTime # 导入列类和关联 from sqlalchemy import Column, ForeignKey

---

# 引擎 uri = 'sqlite:///:memory:' engine = create_engine(uri) # 元信息 meta = MetaData(bind=engine) # 获取数据库类型 db_type = uri.split(':')[0].split('+')[0] # 数据库专业术语 if db_type == 'mysql': # 使用MySQL from sqlalchemy.dialects.mysql.pymysql import dialect # 实例化专业术语对象 dialect_obj = dialect() elif db_type == 'sqlite': # 使用SQLite from sqlalchemy.dialects.sqlite import dialect # 实例化专业术语对象 dialect_obj = dialect() else: # 从引擎中获取专业术语对象 dialect_obj = engine.dialect def SS(*args, **kwargs): global dialect_obj kwargs['dialect_obj'] = dialect_obj print(structure_sql(*args, **kwargs))

---

创建表

# 定义用户表 tb_user = Table( 'users', meta, Column('id', Integer, autoincrement=True, primary_key=True, comment='主键ID'), Column('username', String(20), nullable=False, unique=True, comment='用户名'), Column('password', String(32), nullable=False, comment='密码'), Column('age', Integer, default=18, comment='年龄'), extend_existing=True, comment='用户表' ) # 定义交易表 tb_trade = Table( 'trades', meta, Column('id', Integer, autoincrement=True, primary_key=True, comment='主键ID'), Column('user_id', Integer, ForeignKey('users.id'), nullable=False, comment='用户ID'), Column('amount', Float(2), default=0.0, comment='交易金额'), Column('trade_time', TIMESTAMP, server_default=text('CURRENT_TIMESTAMP'), comment='交易时间'), extend_existing=True, comment='交易表' ) # 创建表 meta.create_all(tables=meta.sorted_tables)

填充数据

# 构造Insert语句 ins = tb_user.insert() # 数据(注意，多条插入时，每一条数据的字段个数和名称需要相同) data = [ {'username': 'swartz', 'password': 'abcdefg', 'age': 26}, {'username': 'gates', 'password': '123456', 'age': 50}, {'username': 'linus', 'password': '123454321', 'age': 50}, {'username': 'bill', 'password': 'abcdefg', 'age': 36}, {'username': 'python', 'password': '666ge666', 'age': 37} ] with engine.connect() as conn: rp_data = conn.execute(ins, data) one = { 'username': 'ruirui', 'password' : '123456' } rp_one = conn.execute(ins, **one) print('添加条数:', rp_data.rowcount + rp_one.rowcount) # 显示插入的数据的id sel_uids = tb_user.select().with_only_columns([tb_user.columns.id]) sel_uids = sel_uids.order_by(tb_user.columns.id.asc()) rp = conn.execute(sel_uids) uids = rp.fetchall() uids = [row[0] for row in uids] rp.close() print(uids) # 添加交易记录 ins_tone = tb_trade.insert() tone = { 'user_id': 1, 'amount': 123.456 } rp_tone = conn.execute(ins_tone, **tone) print('添加', '成功' if rp_tone.rowcount > 0 else '失败') sel_t = tb_trade.select() rp_t = conn.execute(sel_t) t_data = rp_t.fetchall() rp_t.close() print(t_data)

操作

添加、查询数据练习

# 定义模型（创建模型类） class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column('username', String(20), nullable=False) age = Column(Integer, default=18) # 在数据库中创建模型对应的表 Base.metadata.create_all(engine) # 添加一条 obj = User() obj.name = 'ruirui' session.add(obj) # 刷新数据到数据库（未保存） session.flush() # 添加多条 users = [] for username in 'zhangsan,lisi,wangwu,zhaoliu'.split(','): u_obj = User() u_obj.name = username users.append(u_obj) session.add_all(users) # 提交操作到数据库 session.commit() # 建立查询 q = session.query(User) # 显示数据条数 print(q.count()) # 高效显示数据条数的方式（需要使用func.count()，先导入func） from sqlalchemy import func # 设置label，然后属性取值的方式 total = session.query(func.count(User.id).label('total')).first().total # 查询单个值方式 total = session.query(func.count(User.id)).scalar() # 第一条数据 u_f = q.first() # 查看 u_f 的类型 print('是否是User?:', isinstance(u_f, User)) # 所有数据 u_a = q.all() # 指定id的数据（id为3） u_id_3 = q.get(3) # ########## 省略操作 5、6 ##########

销毁引擎

engine.dispose()

转载于:https://www.cnblogs.com/cp9648/p/10415824.html