程式師世界 >> 數據庫知識 >> MYSQL數據庫 >> MySQL綜合教程 >> Python 操作 MySQL 之 pysql 與 ORM(轉載)，pysqlorm

Python 操作 MySQL 之 pysql 與 ORM(轉載)，pysqlorm

編輯：MySQL綜合教程

Python 操作 MySQL 之 pysql 與 ORM(轉載)，pysqlorm

本文針對 Python 操作 MySQL 主要使用的兩種方式講解：

原生模塊 pymsql
ORM框架 SQLAchemy

本章內容：

pymsql 執行 sql 增\刪\改\查語句
pymsql 獲取查詢內容、獲取自增 ID
pymsql 游標
pymsql 更改 fetch 數據類型
pymsql 利用 with 簡化操作
ORM 下載安裝
ORM 史上最全操作

一、pymsql

pymsql 是 Python 中操作 MySQL 的原生模塊，其使用方法和 MySQL 的SQL語句幾乎相同

1、下載安裝

pip3 install pymysql

2、執行SQL

執行 SQL 語句的基本語法：

需要注意的是：創建鏈接後，都由游標來進行與數據庫的操作，當然，拿到數據也靠游標

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
# 創建連接
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
# 創建游標
cursor = conn.cursor()
  
# 執行SQL，並返回收影響行數
effect_row = cursor.execute("update hosts set host = '1.1.1.2'")
  
# 執行SQL，並返回受影響行數
#effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,))
  
# 執行SQL，並返回受影響行數
#effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
  
  
# 提交，不然無法保存新建或者修改的數據
conn.commit()
  
# 關閉游標
cursor.close()
# 關閉連接
conn.close()

3、獲取新創建數據自增ID

可以獲取到最新自增的ID，也就是最後插入的一條數據ID

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor = conn.cursor()
cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
conn.commit()
cursor.close()
conn.close()
  
# 獲取最新自增ID
new_id = cursor.lastrowid

4、獲取查詢數據

獲取查詢數據的三種方式：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor = conn.cursor()
cursor.execute("select * from hosts")
  
# 獲取第一行數據
row_1 = cursor.fetchone()
  
# 獲取前n行數據
# row_2 = cursor.fetchmany(3)

# 獲取所有數據
# row_3 = cursor.fetchall()
  
conn.commit()
cursor.close()
conn.close()

5、移動游標

操作都是靠游標，那對游標的控制也是必須的

注：在fetch數據時按照順序進行，可以使用cursor.scroll(num,mode)來移動游標位置，如：

cursor.scroll(1,mode='relative')  # 相對當前位置移動
cursor.scroll(2,mode='absolute')  # 相對絕對位置移動

6、fetch數據類型

默認拿到的數據是小括號，元祖類型，如果是字典的話會更方便操作，那方法來了：

# 關於默認獲取的數據是元祖類型，如果想要或者字典類型的數據，即：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  
# 游標設置為字典類型
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
  
result = cursor.fetchone()
  
conn.commit()
cursor.close()
conn.close()

7、利用 with 自動關閉

每次連接數據庫都需要連接和關閉，啊，好多代碼，那麼方法又來了：

是不是很屌啊？

# 利用with定義函數

    @contextlib.contextmanager
    def mysql(self, host='127.0.0.1', port=3306, user='nick', passwd='', db='db1', charset='utf8'):
        self.conn = pymysql.connect(host=host, port=port, user=user, passwd=passwd, db=db, charset=charset)
        self.cuersor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)

        try:
            yield self.cuersor
        finally:
            self.conn.commit()
            self.cuersor.close()
            self.conn.close()

# 執行
with mysql() as cuersor:
   print(cuersor)
   # 操作MySQL代碼塊

二、SQLAlchemy

SQLAlchemy 簡稱 ORM 框架，該框架建立在數據庫的 API 之上，使用關系對象映射來進行數據庫操作；

簡言之便是：將類對象轉換成 SQL 語句，然後使用數據 API 執行 SQL 語句並獲取執行結果。

1、下載安裝

pip3 install SQLAlchemy

需要注意了：SQLAlchemy 自己無法操作數據庫，必須結合 pymsql 等第三方插件，Dialect 用於和數據 API 進行交互，根據配置文件的不同調用不同的數據庫 API，從而實現對數據庫的操作，如：

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
   
更多詳見：http://docs.sqlalchemy.org/en/latest/dialects/index.html

2、內部處理

使用 Engine/ConnectionPooling/Dialect 進行數據庫操作，Engine使用ConnectionPooling連接數據庫，然後再通過Dialect執行SQL語句。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
  
  
engine = create_engine("mysql+pymysql://root:[email protected]:3306/t1", max_overflow=5)
  
# 執行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)"
# )
  
# 新插入行自增ID
# cur.lastrowid
  
# 執行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[('1.1.1.22', 3),('1.1.1.221', 3),]
# )
  
  
# 執行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host='1.1.1.99', color_id=3
# )
  
# 執行SQL
# cur = engine.execute('select * from hosts')
# 獲取第一行數據
# cur.fetchone()
# 獲取第n行數據
# cur.fetchmany(3)
# 獲取所有數據
# cur.fetchall()

3、ORM功能使用

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有組件對數據進行操作。

根據類創建對象，對象轉換成SQL，執行SQL。

a、創建表

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

# 指定字符集、最大連接池數
engine = create_engine("mysql+pymysql://root:[email protected]:3306/suoning4?charset=utf8", max_overflow=5)

Base = declarative_base()

# 創建單表
class Users(Base):
    # 表名
    __tablename__ = 'users'
    # 表字段
    id = Column(Integer, primary_key=True)  # 主鍵、默認自增
    name = Column(String(32))
    extra = Column(String(16))

    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'), # 唯一索引
    Index('ix_id_name', 'name', 'extra'),   # 普通索引
    )

    def __repr__(self):
        # 查是輸出的內容格式，本質還是對象
        return "%s-%s" %(self.id, self.name)

# 一對多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True) # 默認值、唯一索引

    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 與生成表結構無關，僅用於查詢方便
    favor = relationship("Favor", backref='pers')

# 多對多
class ServerToGroup(Base):
    # 關系表要放對應表上面，否則找不到
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))    # 外鍵
    group_id = Column(Integer, ForeignKey('group.id'))

    group = relationship("Group", backref='s2g')
    server = relationship("Server", backref='s2g')

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False) # 不能為空
    port = Column(Integer, default=22)
    # group = relationship('Group',secondary=ServerToGroup,backref='host_list')


class Server(Base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True) # 自增
    hostname = Column(String(64), unique=True, nullable=False)


def init_db():
    # 創建表
    Base.metadata.create_all(engine)

def drop_db():
    # 刪除表
    Base.metadata.drop_all(engine)

創建表

注：設置外檢的另一種方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])

b、操作表

操作表那必須導入模塊，創建相應類，相應增\刪\改\查的語法，詳細見下code吧^^：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

# 指定字符集、最大連接池數
engine = create_engine("mysql+pymysql://root:[email protected]:3306/suoning4?charset=utf8", max_overflow=5)

Base = declarative_base()

# 創建單表
class Users(Base):
    # 表名
    __tablename__ = 'users'
    # 表字段
    id = Column(Integer, primary_key=True)  # 主鍵、默認自增
    name = Column(String(32))
    extra = Column(String(16))

    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'), # 唯一索引
    Index('ix_id_name', 'name', 'extra'),   # 普通索引
    )

    def __repr__(self):
        # 查是輸出的內容格式，本質還是對象
        return "%s-%s" %(self.id, self.name)

# 一對多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True) # 默認值、唯一索引

    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 與生成表結構無關，僅用於查詢方便
    favor = relationship("Favor", backref='pers')

# 多對多
class ServerToGroup(Base):
    # 關系表要放對應表上面，否則找不到
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))    # 外鍵
    group_id = Column(Integer, ForeignKey('group.id'))

    group = relationship("Group", backref='s2g')
    server = relationship("Server", backref='s2g')

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False) # 不能為空
    port = Column(Integer, default=22)
    # group = relationship('Group',secondary=ServerToGroup,backref='host_list')


class Server(Base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True) # 自增
    hostname = Column(String(64), unique=True, nullable=False)


def init_db():
    # 創建表
    Base.metadata.create_all(engine)

def drop_db():
    # 刪除表
    Base.metadata.drop_all(engine)


# 先實例化sessionmaker類，Session對象加括號執行類下的__call__方法，
# 得到session對象，所以session可以調用Session類下的add，add_all等方法
Session = sessionmaker(bind=engine) # 指定引擎
session = Session()

# 增

# 添加一條
obj = Users(name="張三", extra='三兒')
session.add(obj)
# 添加多條
session.add_all([
    Users(name="李四", extra='四兒'),
    Users(name="汪五", extra='五兒'),
])
# 提交
session.commit()

# 刪

session.query(Users).filter(Users.id > 2).delete()
session.query(Users).filter_by(id = 1).delete()
session.commit()

# 改

session.query(Users).filter(Users.id > 2).update({"name" : "nick"})
session.query(Users).filter(Users.id > 2).update({"name" : "nick", "extra":"niubily"})
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "Suo"}, synchronize_session=False)
session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
session.commit()

# 查
# all()結果為對象列表，first()為具體對象

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

那如何加限制條件等，我要更靈活使用，好吧，還是見下 code：

#　條件
ret = session.query(Users).filter_by(name='nick').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'nick').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'nick').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='nick'))).all()

from sqlalchemy import and_, or_        # 導入模塊
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'nick')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'nick')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'nick', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like('n%')).all()
ret = session.query(Users).filter(~Users.name.like('n%')).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()

# isouter=True 理解為 left join ，如果不寫為 inner join

# 組合
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()