索引

批量录入数据

当需要插入大量数据的时候，如果依然采用逐个插入的方法，那么就会在和数据库的交互上浪费很多时间，效率很低。MySQL 等大多数数据库都提供了 insert ... values (...), (...) ... 这种批量插入的 API，在 SQLAlchemy 中也可以很好地利用这一点。

# 使用 session.bulk_save_objects(...) 直接插入多个对象

s = Session()
objects = [
    User(name="u1"),
    User(name="u2"),
    User(name="u3")
]
s.bulk_save_objects(objects)
s.commit()

# 使用 bulk_insert_mappings 可以省去创建对象的开销，直接插入字典
users = [
    {"name": "u1"},
    {"name": "u2"},
    {"name": "u3"},
]
s.bulk_insert_mappings(User, users)
s.commit()

# 使用 bulk_update_mappings 可以批量更新对象，字典中的 id 会被用作 where 条件，
# 其他字段全部用于更新
session.bulk_update_mappings(User, users)

select语句

事务

模型类中自定义特殊方法

对于查询结果来说，如果返回的是模型类对象，如果模型类中，不写__repr__方法，则默认返回模型类对象，写了__repr__方法，可以更友好的返回一些字符串，便于展示。

我们也可以在模型类中，自定义一些方法，实现一些特殊的功能：

比如处理明文手机号18211101111为 182****1111。
获取明文密码123对应的加密后的密文密码936cb9f4e2db06ed32d75d57d63f9c3d。
也可以指定返回列表套字典的形式[{'name': '刘桂芝', 'birth': '2023-09-14 21:55:47', 'money': 70.07, 'mobile': '13027918506', 'encrypt_mobile': '130****8506', 'pwd': '123', 'md5_pwd': '936cb9f4e2db06ed32d75d57d63f9c3d'}]。

python

# -*- coding = utf-8 -*-
import random
import string
import faker  # https://www.cnblogs.com/Neeo/articles/11316724.html
from hashlib import md5
from datetime import datetime
from sqlalchemy import URL
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import sqlalchemy as db
from sqlalchemy.orm import declarative_base
from sqlalchemy.dialects import mysql

url = URL.create(drivername="mysql", username='root', password='123', host='127.0.0.1', port=3306, database='alchemy')
engine = create_engine(url=url)
Session = sessionmaker(bind=engine)
session = Session()
Model = declarative_base()

class Test(Model):
    __tablename__ = "tb_test"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment='主键ID')
    name = db.Column(db.String(20), comment='用户名')
    money = db.Column(db.DECIMAL(precision=10, scale=2), comment='书籍价格')
    mobile = db.Column(db.String(20), comment='手机号')
    pwd = db.Column(db.String(20), comment='密码')
    birth = db.Column(db.DateTime, default=datetime.now, comment='出生时间')

    def __repr__(self):
        """
        当实例对象被使用print打印时，自动执行该__repr__方法, 返回值必须时字符串格式，否则报错！！！
        """
        return f"<{self.name}>"

    @property
    def to_dict1(self):
        """ 这种方式比较灵活，想返回啥返回啥 """
        return {
            'name': self.name,
            'birth': self.birth.strftime('%Y-%m-%d %H:%M:%S'),
            'money': float(self.money),  # 对于DECIMAL类型无法别json直接序列化，我们这里也可以提前进行转成float类型
            'money2': f"¥{float(self.money)}",  # 也可以按需进行预处理
            'mobile': self.mobile,
            'encrypt_mobile': self.get_encrypt_mobile,
            'pwd': self.pwd,
            'md5_pwd': self.get_md5_pwd
        }

    @property
    def to_dict2(self):
        """ 这种方式就比较简单粗暴了，直接返回当前模型类对象的所有字段 """
        dict_ = self.__dict__
        if "_sa_instance_state" in dict_:
            del dict_["_sa_instance_state"]
        return dict_

    @property
    def get_encrypt_mobile(self):
        return f"{self.mobile[:3]}****{self.mobile[-4:]}"

    @property
    def get_md5_pwd(self):
        salt = 'qwe1qwp[];/;83421d/mfsdf9p[had123'
        return md5(f'{salt}{self.pwd}'.encode('utf8')).hexdigest()


if __name__ == '__main__':
    # Model.metadata.drop_all(engine)
    # Model.metadata.create_all(engine)
    # fk = faker.Faker(locale='zh_CN')
    # # 插入数据
    # test_objs = [
    #     Test(
    #         name=fk.name(), money=round(random.random() * 100, 2),
    #         mobile=f'130{"".join(random.sample(string.digits, 8))}',
    #         pwd='123'
    #     )
    #     for i in range(3)
    # ]
    # session.add_all(test_objs)
    # session.commit()

    # 如果模型类中，不写__repr__方法，则默认返回模型类对象，写了__repr__方法，可以更友好的返回一些字符串，便于展示
    # 我们也可以在模型类中，自定义一些方法，实现一些特殊的功能
    # 比如处理明文手机号为 182****1111;获取明文密码对应的密文密码;也可以指定返回列表套字典的形式
    objs = session.query(Test).limit(1).all()
    print(objs)  # [<刘桂芝>]
    print([obj.get_md5_pwd for obj in objs])  # ['936cb9f4e2db06ed32d75d57d63f9c3d']
    print([obj.get_encrypt_mobile for obj in objs])  # ['130****8506']
    print([obj.to_dict1 for obj in objs])  # [{'name': '刘桂芝', 'birth': '2023-09-14 21:55:47', 'money': 70.07, 'money2': '¥70.07', 'mobile': '13027918506', 'encrypt_mobile': '130****8506', 'pwd': '123', 'md5_pwd': '936cb9f4e2db06ed32d75d57d63f9c3d'}]
    print([obj.to_dict2 for obj in objs])  # [{'name': '刘桂芝', 'money': Decimal('70.07'), 'pwd': '123', 'mobile': '13027918506', 'id': 1, 'birth': datetime.datetime(2023, 9, 14, 21, 55, 47)}]

索引 ​

批量录入数据 ​

select语句 ​

事务 ​

模型类中自定义特殊方法 ​

索引

批量录入数据

select语句

事务

模型类中自定义特殊方法