Python手册(8) MySQL

发表于 | 分类于 ，
字数统计: 994 | 阅读时长 ≈ 4
介绍Python中MySQL的基本使用（SQLAlchemy包）

0. 前言

  • 安装MySQL驱动：pip install pymysql
  • 安装ORM框架：pip install sqlalchemy
  • SQLAlchemy资料：

1. SQLAlchemy 的基本使用

1.1. 创建engine

  • 主要两个功能：
    1. 构建MySQL的URL，格式为数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名
    2. 在URL中处理编码问题，具体参考 这里。
      1
      2
      3
      4
      from sqlalchemy import create_engine

      # echo如果为True，则后续操作会输出很多日志，看需求
      engine = create_engine("mysql+pymysql://root:123456@0.0.0.0/db_name?charset=utf8", encoding='utf8', echo=True)

1.2. 创建表映射 & 新建表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float

Base = declarative_base()

# 表结构
class MyTable(Base):
    __tablename__ = 't_table_name'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32))
    unit_price = Column(Float)
    address = Column(String(255))
    comment = Column(String(255))

# 新建表
Base.metadata.create_all(engine)

1.3. 建立数据库记录实例

1
record = MyTable(name="XX小区", unit_price=22000.5, address="中山路50号", comment="其他信息")

1.4. 创建会话实例

  • 提供了两种方式进行操作，下面分别介绍。

  • 使用Session实例，通过ORM或SQL语句的方式操作数据库。

    • session操作要通过commit提交。在使用完session实例后，要调用close关闭。
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      # 创建Session实例
      from sqlalchemy.orm import sessionmaker
      Session = sessionmaker(bind=engine)
      session = Session()

      # Insert操作
      # 流程：先建立数据库记录实例，再通过`session`的`add`添加一条记录，通过`add_all`实现批量添加。
      record = MyTable(name="XX", unit_price=22000.5, address="50", comment="other")
      session.add(record)
      session.add_all([MyTable(name="XXX小区", unit_price=3000.5, address="建国路50号", comment="其他信息"),
                      MyTable(name="XXXX小区", unit_price=41000.5, address="和平路50号", comment="其他信息"),
                      MyTable(name="XXXXX小区", unit_price=13000.5, address="天目山路50号", comment="其他信息")])
      session.commit()

      # Select操作
      # 流程：通过session的`query`获取查询结果，对结果通过`filter`筛选，得到数据库记录实例。
      q = session.query(MyTable)
      q = q.filter(MyTable.id==5)
      record = q.one()
      print(record.id, record.name, record.address, record.comment)

      # Update操作
      # 流程：先通过Select操作获取数据库记录实例，修改实例变量，再提交。
      record.name = 'UPDATE NAME'
      session.commit()

      # Delete操作
      # 流程：先通过Select获取数据库实例，再通过`session`的`delete`函数删除。
      session.delete(record)
      session.commit()

      # 直接执行SQL语句
      session.execute('SELECT * FROM t_user')

  • 使用Connection实例，直接通过SQL语句操作数据库。

    1
    2
    3
    4
    # 直接执行sql语句
    with engine.connect() as con:
        rs = con.execute('SELECT * FROM t_user')
        data = rs.fetchone()[0]

2. 小功能

2.1. 自动设置 DateTime 相关字段

  • 主要参考这篇博客
  • 注意 onupdate 字段，具体参考下面代码
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    from uuid import uuid4
    from sqlalchemy import Column, Integer, String, DateTime, Boolean, TIMESTAMP, func
    from sqlalchemy.orm import relationship
    from sqlalchemy_demo.connect import Base

    class UserModule(Base):
        __tablename__ = 'user'

        uuid = Column(String(36), unique=True, nullable=False, default=lambda: str(uuid4()), comment='uuid')
        id = Column(Integer, primary_key=True, autoincrement=True, comment='用户id')
        user_name = Column(String(30), nullable=False, unique=True, comment='用户名')
        password = Column(String(64), nullable=False, comment='用户密码')
        createtime = Column(DateTime, server_default=func.now(), comment='创建时间')
        # onupdate设置自动更改
        updatetime = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment='修改时间')

3. 实例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker

MYSQL_URL = "mysql+pymysql://root:123456@0.0.0.0/db_name?charset=utf8"
MYSQL_ECHO = True  # 如果为True，则后续操作会输出很多日志，看需求

# 创建引擎以及相关内容
engine = create_engine(MYSQL_URL, encoding='utf8', echo=MYSQL_ECHO)
Base = declarative_base()


# 定义表结构（表名、字段）
class MyTable(Base):
    __tablename__ = 't_table_name'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32))
    unit_price = Column(Float)
    comment = Column(String(255))


# 新建表
Base.metadata.create_all(engine)


# 通过Session，以ORM的形式来增删改差
# 创建Session实例
Session = sessionmaker(bind=engine)
session = Session()
# 增
record = MyTable(name="XX", unit_price=22000.5, comment="other")
session.add(record)
session.add_all([MyTable(name="XXX小区", unit_price=3000.5, comment="其他信息1"),
                 MyTable(name="XXXX小区", unit_price=41000.5, comment="其他信息2"),
                 MyTable(name="XXXXX小区", unit_price=13000.5, comment="其他信息3")])
session.commit()
# 查
q = session.query(MyTable)
q = q.filter(MyTable.id==5)
record = q.one()
print(record.id, record.name, record.address, record.comment)
# 改
record.name = 'UPDATE NAME'
session.commit()
# 删
session.delete(record)
session.commit()
# 通过Session直接执行sql语句
session.execute('SELECT * FROM t_user')


# 通过connection执行sql语句
with engine.connect() as con:
    rs = con.execute('SELECT * FROM t_user')
    data = rs.fetchone()[0]