【轉】MySQL— pymysql and SQLAlchemy Python開發【第十九篇】：Python操作MySQL
阿新 • • 發佈：2018-12-05
【轉】MySQL— pymysql and SQLAlchemy
目錄
一、pymysql
pymsql是Python中操作MySQL的模組，其使用方法和MySQLdb幾乎相同。
1. 下載安裝
#在終端直接執行
pip3 install pymysql
2. 使用操作
a. 執行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()
b. 獲取新建立資料自增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() # 獲取最新自增ID new_id = cursor.lastrowid cursor.close() conn.close()
c. 獲取查詢資料
#!/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()
注：在fetch資料時按照順序進行，可以使用cursor.scroll(num,mode)來移動遊標位置，如：
- cursor.scroll(1,mode='relative') # 相對當前位置移動
- cursor.scroll(2,mode='absolute') # 相對絕對位置移動
d. 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()
二、SQLAlchemy
SQLAlchemy是Python程式語言下的一款ORM框架，該框架建立在資料庫API之上，使用關係物件對映進行資料庫操作，簡言之便是：將物件轉換成SQL，然後使用資料API執行SQL並獲取執行結果。
1. 下載安裝
#在終端直接執行 pip3 install SQLAlchemy
2. 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
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 #表明依賴關係並建立連線，最大連線數為5 engine = create_engine("mysql+pymysql://root:[email protected]:3306/t1", max_overflow=5) Base = declarative_base() # 建立單表 class Users(Base): __tablename__ = 'users' # 表名 id = Column(Integer, primary_key=True,autoincrement=True) # id列，主鍵自增 name = Column(String(32)) # name列 extra = Column(String(16)) # extra列 __table_args__ = ( UniqueConstraint('id', 'name', name='uix_id_name'), # 建立聯合唯一索引 Index('ix_id_name', 'name', 'extra'), # 建立普通索引 ) # 一對多 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")) # 建立外來鍵 # 多對多 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)
注：設定外來鍵的另一種方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])
b. 操作表
#!/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/t1", 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) Session = sessionmaker(bind=engine) session = Session()表結構 + 資料庫連線
b.1 增
#單條增加 obj = Users(name="alex0", extra='sb') session.add(obj) #多條增加 session.add_all([ Users(name="alex1", extra='sb'), Users(name="alex2", extra='sb'), ]) #提交 session.commit()
b.2 刪
#先查詢到要刪除的記錄，再delete session.query(Users).filter(Users.id > 2).delete() session.commit()
b.3 改
#先查詢，再更新 session.query(Users).filter(Users.id > 2).update({"name" : "099"}) # 直接更改 session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False) # 字串拼接 session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate") # 數字相加 session.commit()
b.4 查
ret = session.query(Users).all() ret = session.query(Users.name, Users.extra).all() ret = session.query(Users).filter_by(name='alex').all() ret = session.query(Users).filter_by(name='alex').first() ret = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(User.id).all() ret = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()
b.5 其它
# 條件 ret = session.query(Users).filter_by(name='alex').all() # 條件內為關鍵字表達式 ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() # 條件內為SQL表示式 ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() # between ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() # in ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() # not in 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() # and ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() # or 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() # e開頭 ret = session.query(Users).filter(~Users.name.like('e%')).all() # 非e開頭 # 限制 ret = session.query(Users)[1:2] # 相當於limit # 排序 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() # 預設內連 inner join 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()
參考資料：
1. Python開發【第十九篇】：Python操作MySQL