MJX's Note Book
Series
Appearance

Python操作SQL的两种方式

MJX2022/11/06PythonMysql

pymyql

PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库，废话不多说，直接开始。

安装

pip install pymysql

数据库连接

pymysql 提供 connect() 用于数据库连接。我们需要指定以常用参数，方便确定数据库的位置以及编码方式：

  • host： Mysql 地址，远程数据库的话用 IP地址
  • user：Mysql 用户名
  • password：密码
  • database：数据库库名
  • port ：Mysql 端口，如果有手动更改过的话， 一般默认是 3306
  • charset：编码方式
#!/usr/bin/env python
# coding: utf-8

import pymysql

host = 'localhost'  # 地址远程的话用 IP
user = 'root'  # Mysql 用户名
password = '123'  # 密码
database = 'Database'  # 库名
dbconnect = pymysql.connect(host=host, port=3306, user=user, password=password, database=database,
                            charset='utf8')

建库建表

pymsql 操作数据库一般有以下三个步骤：

  1. 连接数据库
  2. 创建游标对象
  3. 执行 SQL 语句
  4. 插入、更新、删除操作需要提交
  5. 关闭连接

我们可以使用 cursor() 创建游标对象，execute() 执行 SQL 语句。

import pymysql

dbconnect = pymysql.connect(host='localhost', user='root', password='123', database='studet')
dbcursor = dbconnect.cursor()  # 创建游标

# 使用预处理语句创建表
sql = """CREATE TABLE student (
         NAME  CHAR(20) NOT NULL,
         AGE INT,  
         SEX CHAR(1)"""
dbcursor.execute(sql)   # 执行 sql 语句
dbconnect.close()  # 关闭数据库连接

插入数据

和上面一样，先创建游标对象，再执行 SQL 语句，不过 需要加上 commit。

import pymysql

dbconnect = pymysql.connect(host='localhost', user='root', password='123', database='studet')
dbcursor = dbconnect.cursor()  # 创建游标

# SQL 插入语句
sql = """INSERT INTO student (NAME, AGE, SEX)
         VALUES ('Ma', 20, 'M')"""

cursor.execute(sql)  # 执行sql语句
db.commit()  # 提交到数据库执行
db.close()  # 关闭数据库连接

更新与删除

所有操作和上面都一样的，只需要修改 SQL。千万不要忘记 commit 操作就行，否则 SQL 语句不会生效。

回滚

对于支持事务的数据库， 在 Python 数据库编程中，当游标建立之时，就自动开始了一个隐形的数据库事务。commit() 方法提交游标的所有更新操作，rollback() 方法回滚 当前游标 的所有操作。每一个方法都开启一个新的事务。

import pymysql
 
# 打开数据库连接
db = pymysql.connect(host='localhost',
                     user='testuser',	
                     password='test123',
                     database='TESTDB')
 
# 使用cursor()方法获取操作游标 
cursor = db.cursor()
 
# SQL 删除语句
sql = "DELETE FROM EMPLOYEE WHERE AGE > %s" % (20)
try:
   # 执行SQL语句
   cursor.execute(sql)
   # 提交修改
   db.commit()
except:
   # 发生错误时回滚
   db.rollback()
    
db.close()

总结

不难看看出使用这种方式操作数据库，其实很简单。因为操作数据的重点还是写 SQL 语句。只需要使用函数来执行 SQL 而已。这种方式唯一的难点就是怎么通过参数拼接 SQL 。

但是这种方式如果不对输入进行限制会有比较严重的 SQL注入 风险。

SQL Alchemy

SQLAlchemy 是一个Python语言编写的高性能的关系型数据库ORM框架，它提供了高层的 ORM 和底层的原生数据库的操作。

ORM

ORM 全拼Object-Relation Mapping，中文意为 对象-关系映射。主要实现模型对象到关系数据库数据的映射。

ORM提供了一种面向对象操作数据库的方式给开发者。不需要编写原生SQL语句也能操作数据库，实现了业务代码与底层数据的解耦。

优点：

  • 只需要面向对象编程, 不需要面向数据库编写 SQL。
    • 对数据库的操作都转化成对类/对象的属性和方法的操作。
    • 不用编写各种数据库的原生sql语句，也可以编写原生SQL语句。
  • 实现了数据模型代码与数据库数据的解耦, 屏蔽了不同数据库操作上的差异。
    • 不再需要关注当前项目使用的是哪种数据库。
    • 通过简单的配置就可以轻松更换数据库，而不需要修改业务代码.

缺点：

  • 相比较直接使用SQL语句操作数据库,ORM需要把操作转换成SQL语句，所以有性能损失.
  • 根据对象的操作转换成SQL语句,根据查询的结果转化成模型实例对象, 在映射过程中有性能损失.
  • 不同的ORM提供的操作不一样，增加了学习成本

安装

pip install SQLAlchemy

数据库连接

from sqlalchemy import create_engine  # 数据库引擎

engine = create_engine('mysql+pymysql://root:123@localhost:3306/todo?charset=utf8mb4',
                       echo=True, pool_size=5, pool_recycle=60 * 60)

部分参数及解释：

  • echo：默认为 FALSE，如果为真，将输出转化后的 SQL 语句。
  • encoding：编码，默认为 utf-8。
  • pool_size：连接池大小，默认为5，设置为0时表示无限制。
  • pool_recycle：设置时间以限制数据库多久没连接自动断开。

使用SQL语句操作

sqlalchemy 除了使用面向对象的方式，其实也提供的原生 SQL 操作数据库的方法。

官方参考open in new window

# 方式一、直接使用核心
from sqlalchemy import create_engine  # 数据库引擎
from sqlalchemy import text  # text 函数用于创建一个 SQL 表达式对象

engine = create_engine('mysql+pymysql://root:123@localhost:3306/todo?charset=utf8mb4',
                       echo=True, pool_size=5, pool_recycle=60 * 60)

with engine.connect() as con:
    ans = con.execute(text('SELECT id, title FROM topic'))  
    """
    可以在后面参数中,添加SQL的参数如：
        conn.execute(
        	text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
            [{"x": 11, "y": 12}, {"x": 13, "y": 14}],
        )
    具体看官方参考（上面链接）
    """ 
    print(ans.fetchall())
    # 同样如果是更新删除操作需要提交 conn.commit()

# 方式二、使用会话
from sqlalchemy import create_engine  # 数据库引擎
from sqlalchemy import text
from sqlalchemy.orm import Session  # Session 有多种不同的创建模式

engine = create_engine('mysql+pymysql://root:123@localhost:3306/todo?charset=utf8mb4',
                       echo=True, pool_size=5, pool_recycle=60 * 60)

with Session(engine) as session:
    ans = session.execute(text('SELECT id, title FROM topic'))
    print(ans.fetchall())
    # 更新删除操作需要提交 session.commit()

声明映射（元数据）

就是我们在 Python 中创建一个类，对应着数据库中的一张表，类的每个属性，就是这个表的字段名。这种的类对应于数据库中表的类，就称为映射类，我们要创建一个映射类来申明一张表。

官方参考open in new window

# 方式一、直接构造
from sqlalchemy import MetaData
from sqlalchemy import Table, Column, Integer, String

metadata_obj = MetaData()
user_table = Table(
    "user_account",
    metadata_obj,
    Column("id", Integer, primary_key=True),  # 可以指定主键、外键、默认值、是否为空等约束
    Column("name", String(30)),
    Column("fullname", String(30)),
)

# 方式二、声明式
from sqlalchemy import create_engine  # 数据库引擎
from sqlalchemy import String, Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column

engine = create_engine('mysql+pymysql://root:123@localhost:3306/todo?charset=utf8mb4',
                       echo=True, pool_size=5, pool_recycle=60 * 60)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30))
    fullname = mapped_column(String(30))

    def __repr__(self):
        return f"User(id={self.id}, name={self.name}, fullname={self.fullname})"

# 在 1.4 及以前版本中作用 Column() 而不是 mapped_column()
# mapped_column() 仅适用于 ORM 声明性映射, 它不能在 Table 构造中使用！

申明好表的话就可以通过 user_table 对象使用各种方法操作表了。

插入数据

官方参考open in new window

# 方式一
from sqlalchemy import create_engine
from sqlalchemy import String, Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy import insert

engine = create_engine('mysql+pymysql://root:123@localhost:3306/todo?charset=utf8mb4',
                       echo=True, pool_size=5, pool_recycle=60 * 60)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30))
    fullname = mapped_column(String(30))

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

# 如果事先没有创建表需要在下面语句创建一下
# Base.metadata.create_all(engine)

ins_SQL = insert(User).values(id=1, name='M', fullname='MJX')
print(ins_SQL)
with engine.connect() as conn:
    result = conn.execute(ins_SQL)
    conn.commit()

# 方式二（推荐）
from sqlalchemy import create_engine
from sqlalchemy import String, Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import Session  # Session 有多种不同的创建

engine = create_engine('mysql+pymysql://root:123@localhost:3306/todo?charset=utf8mb4',
                       echo=True, pool_size=5, pool_recycle=60 * 60)

class Base(DeclarativeBase):
    pass

class User(Base):
    
    __tablename__ = "user_account"

    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String(30))
    fullname = mapped_column(String(30))

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

user = User(id=2, name='M', fullname='MJX')
with Session(engine) as session:
    ans = session.add(user)
    session.commit()

查询数据

官方参考open in new window

查询语句和上面的相差其实就不大了。

# 方式一
from sqlalchemy import select
stmt = select(user_table).where(user_table.c.name == "spongebob")
print(stmt)

with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(row)
        
# 方式二
with Session(engine) as session:
    ans = session.query(User).filter_by(name='ed')
Edit this page
ON THIS PAGE