(目录)

---

一、如何获取新插入数据的自增ID

from pymysql import Connection

conn = None
try:
    # 创建数据库连接
    conn = Connection(
        host="localhost",  # 主机名
        port=3306,  # 端口
        user="root",  # 账户
        password="123456",  # 密码
        database="test",  # 指定操作的数据库
        autocommit=True,  # 设置自动提交
        charset="utf8mb4",  # 设置字符集
    )
    with conn.cursor() as cursor:  # 上下文管理器，自动关闭游标
        # 创建一个带有占位符的SQL语句——这里的占位符是%s
        sql = "insert into `test_info` (`id`, `u_name`, `u_tel`, `u_mark`) values (default, %s, %s, %s)"
        # 执行SQL语句，传入参数。注意，参数是通过execute()方法的第二个参数传入的。
        data = ("洪七公", "13888666666", "老叫花子")
        cursor.execute(sql, data)

        # 获取插入最后一行的ID（需在conn.commit()提交之后）
        id = cursor.lastrowid
        print("插入成功，ID:", id)

except Exception as e:
    conn.rollback()  # 回滚事务
    print("异常：", e)
finally:
    if conn:
        # 关闭连接
        conn.close()

二、ORM框架SQLAlchemy

安装软件包

pip install SQLAlchemy

参考：https://www.cnblogs.com/wupeiqi/articles/5713330.html 官方文档：https://docs.sqlalchemy.org/en/20/dialects/index.html

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作。

1. 创建表

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, UniqueConstraint, Index

Base = declarative_base()


class UserType(Base):
    __tablename__ = "usertype"
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(32), nullable=True, index=True)


# 创建单表
class Users(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=True, default=None, index=True)
    email = Column(String(16), unique=True)
    user_type_id = Column(Integer, ForeignKey="usertype.id")

    __table_args__ = (
        UniqueConstraint(
            "id", "name", name="uix_id_name"
        ),  # 注意位置参数不能出现在关键字参数之后
        Index("ix_id_name", "name", "email"),
    )


engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8mb4", max_overflow=5
)
# 将上边所有继承了Base的类按数据库表创建出来
Base.metadata.create_all(engine)

• 连接数据库是通过pymysql
• SQLAlchemy负责对象到SQL语句的转换
• max_overflow=5表示可以同时并发5个连接
• Base.metadata.drop_all(engine) 删除所有的表

2. 增删改查

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8mb4", max_overflow=5
)

Session = sessionmaker(bind=engine)
session = Session()

# 类->表
# 对象->行
##### 增
obj1 = UserType(title="admin")
session.add(obj1)

objs = [
    UserType(title="root"),
    UserType(title="user"),
    UserType(title="guest"),
]
session.add_all(objs)

##### 查
user_type_list = session.query(UserType).all()
for user_type in user_type_list:
    print(user_type.id, user_type.title)

user_type_list = session.query(UserType.id, UserType.title).filter(UserType.id > 3)
for user_type in user_type_list:
    print(user_type.id, user_type.title)

##### 删
session.query(UserType.id, UserType.title).filter(UserType.id > 30).delete()

##### 改
session.query(UserType.id, UserType.title).filter(UserType.id < 3).update(
    {"title": "new_title"}
)

session.query(UserType).filter(UserType.id > 2).update(
    {UserType.title: UserType.title + "er"}, synchronize_session=False
)

session.query(UserType).filter(UserType.id > 2).update(
    {"level": UserType.level + 1}, synchronize_session="evaluate"
)

session.commit()
session.close()

3. 各种常见的花式查询

ret = session.query(Users).all()
ret = session.query(Users.name, Users.extra).all()
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter_by(name='alex').first()

ret = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(User.id).all()

ret = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()

# 子查询
q1 = session.query(UserType).filter(UserType.id > 2).subquery()
result = session.query(q1).all()

4. 其他常用技巧

#　条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()