Python操作Mysql

本文針對(duì)Python操作Mysql的兩種方式

  • 原生模塊
  • ORM框架 SQLAchemy

Pymsql

pymsql是Python中操作MySQL的模塊，其使用方法和MySQLdb幾乎相同。

下載安裝

pip3 install pymysql
使用操作

1、執(zhí)行sql語(yǔ)句

import pymysql

# 創(chuàng)建連接
# host mysql所在的IP地址
# user password    用戶名、密碼
# database  操作的數(shù)據(jù)庫(kù)
conn = pymysql.connect(host = "114.116.51.xxx", user='root', password='root123', database='DB_Rainy')

# 創(chuàng)建游標(biāo)
cursor = conn.cursor()

# 執(zhí)行SQL，并返回收影響行數(shù)
effect_row = cursor.execute("update hosts set host = '1.1.1.2'")
  
# 執(zhí)行SQL，并返回受影響行數(shù)
effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,))
  
# 執(zhí)行SQL，并返回受影響行數(shù)
effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
  
  
# 提交，不然無(wú)法保存新建或者修改的數(shù)據(jù)
# 查詢不需要commeit，修改、插入、刪除都需要commit
conn.commit()
  
# 關(guān)閉游標(biāo)
cursor.close()
# 關(guān)閉連接
conn.close()

2、獲取新創(chuàng)建數(shù)據(jù)自增ID

import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', 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

3、獲取查詢數(shù)據(jù)

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

注意：在fetch數(shù)據(jù)時(shí)按照順序進(jìn)行，可以使用cursor.scroll(num,mode)來(lái)移動(dòng)游標(biāo)位置，如：

  • cursor.scroll(1,mode='relative') # 相對(duì)當(dāng)前位置移動(dòng)
  • cursor.scroll(2,mode='absolute') # 相對(duì)絕對(duì)位置移動(dòng)

4、fetch數(shù)據(jù)類(lèi)型
　　關(guān)于默認(rèn)獲取的數(shù)據(jù)是元祖類(lèi)型，如果想要或者字典類(lèi)型的數(shù)據(jù)，即:

import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  
# 游標(biāo)設(shè)置為字典類(lèi)型
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
  
result = cursor.fetchone()
  
conn.commit()
cursor.close()
conn.close()

SQLAchemy

SQLAlchemy是Python編程語(yǔ)言下的一款ORM框架，該框架建立在數(shù)據(jù)庫(kù)API之上，使用關(guān)系對(duì)象映射進(jìn)行數(shù)據(jù)庫(kù)操作，簡(jiǎn)言之便是：將對(duì)象轉(zhuǎn)換成SQL，然后使用數(shù)據(jù)API執(zhí)行SQL并獲取執(zhí)行結(jié)果。

安裝


image.png
pip3 install SQLAlchemy

SQLAlchemy本身無(wú)法操作數(shù)據(jù)庫(kù)，其必須以來(lái)pymsql等第三方插件，Dialect用于和數(shù)據(jù)API進(jìn)行交流，根據(jù)配置文件的不同調(diào)用不同的數(shù)據(jù)庫(kù)API，從而實(shí)現(xiàn)對(duì)數(shù)據(jù)庫(kù)的操作，如：

# 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...]
   
# 更多詳見(jiàn)：http://docs.sqlalchemy.org/en/latest/dialects/index.html

1、內(nèi)部處理

使用 Engine/ConnectionPooling/Dialect 進(jìn)行數(shù)據(jù)庫(kù)操作，Engine使用ConnectionPooling連接數(shù)據(jù)庫(kù)，然后再通過(guò)Dialect執(zhí)行SQL語(yǔ)句。

from sqlalchemy import create_engine
  
  
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  
# 執(zhí)行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)"
# )
  
# 新插入行自增ID
# cur.lastrowid
  
# 執(zhí)行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[('1.1.1.22', 3),('1.1.1.221', 3),]
# )
  
  
# 執(zhí)行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host='1.1.1.99', color_id=3
# )
  
# 執(zhí)行SQL
# cur = engine.execute('select * from hosts')
# 獲取第一行數(shù)據(jù)
# cur.fetchone()
# 獲取第n行數(shù)據(jù)
# cur.fetchmany(3)
# 獲取所有數(shù)據(jù)
# cur.fetchall()

2、ORM功能使用
使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有組件對(duì)數(shù)據(jù)進(jìn)行操作。根據(jù)類(lèi)創(chuàng)建對(duì)象，對(duì)象轉(zhuǎn)換成SQL，執(zhí)行SQL。

2.1 創(chuàng)建表

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:root123@127.0.0.1:3306/t1", max_overflow=5)
 
Base = declarative_base()
 
# 創(chuàng)建單表
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'),
    )
 
# 一對(duì)多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)
 
 
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"))
 
 
# 多對(duì)多
class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
 
 
class Server(Base):
    __tablename__ = 'server'
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
 
 
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'))
 
 
def init_db():
    Base.metadata.create_all(engine)
 
 
def drop_db():
    Base.metadata.drop_all(engine)

創(chuàng)建外鍵的另外一種方式

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:root123@114.116.51.242:3306/db1",max_overflow=5)

Base = declarative_base()

class User(Base):
    __tablename__ = 'userinfo'
    id = Column(Integer,autoincrement=True,primary_key=True)
    name = Column(String(20))
    email = Column(String(32))
    num = Column(Interger)

    __table_args__ = (
        UniqueConstraint("name",name='uix_n'),
        Index('Ix_e','email')
    )

class passwd(Base):
    __tablename__ = 'password'
    id = Column(Integer,autoincrement=True,primary_key=True)
    pwd = Column(String(32))
    userid = Column(Integer,ForeignKey('userinfo.id'),default='123')
    
def init_db():
    Base.metadata.create_all(engine)

def drop_db():
    Base.metadata.drop_all(engine)

2.2 操作表

增刪改查

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:root123@114.116.51.242:3306/db1",max_overflow=5)

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

# 增加數(shù)據(jù)
"""
# 插入一行數(shù)據(jù)
obj = User(name='Wang',email='Wang@163.com')

session.add(obj)
session.commit()
session.close()

# 插入多行數(shù)據(jù)
objs = [
    User(name="Rainy1",email="13620606694@163.com"),
    User(name="Rainy2",email="13620606694@163.com"),
    User(name="Rainy3",email="13620606694@163.com"),
]

session.add_all(objs)
session.commit()
session.close()

"""


# 查詢
'''
# 查詢?nèi)繑?shù)據(jù)
userinfo = session.query(User).all()
for i in userinfo:
    print(i.id,i.name,i.email)

# 按條件查詢
userinfo = session.query(User).filter(User.name == 'Rainy')

for i in userinfo:
    print(i.id,i.name,i.email)

# 查詢指定列
userinfo = session.query(User.name,User.email).filter(User.name == 'Wang')
for i in userinfo:
    print(i.name,i.email)
'''

# 刪除
'''
# 刪除指定行<先查詢出來(lái)后直接刪除>
info = session.query(User).filter(User.name == 'Wang').delete()
session.commit()
session.close()

# 刪除全部數(shù)據(jù)
    查詢出全部數(shù)據(jù)后后面直接delete()
'''

# 修改
'''
# 修改指定行數(shù)據(jù)的指定列修改為指定值
info = session.query(User).filter(User.name == "Wang").update({'email':'123@123.com'})
session.commit()
session.close()

# 對(duì)指定行的指定列 拼接字符串
info = session.query(User).filter(User.id < 6).update({User.name:User.name + 'Wang'},synchronize_session=False)
session.commit()
session.close()

# 對(duì)指定行對(duì)指定列中的數(shù)字進(jìn)行處理
info = session.query(User).filter(User.id < 3).update({'num': User.num + 1},synchronize_session="evaluate")
session.commit()
session.close()
'''

其他

#　條件
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()
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容