sqlalchemy 流程演示

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))

---

1. 创建表

   # 定义用户表
   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)
   

2. 填充数据

   # 构造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)
   

3. 操作

   添加、查询数据练习

   # 定义模型（创建模型类）
   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 ##########
   

4. 销毁引擎

   engine.dispose()