一、環境依賴
1、安裝mysqlclient子產品
#下載下傳mysqlclient包
進入https://www.lfd.uci.edu/~gohlke/pythonlibs/#mysqlclient進行下載下傳
#在對應的虛拟環境中進行安裝
(automatic) C:\Users\Administrator>pip install E:\python\mysqlclient-1.4.2-cp35-
cp35m-win_amd64.whl
Processing e:\python\mysqlclient-1.4.2-cp35-cp35m-win_amd64.whl
Installing collected packages: mysqlclient
Successfully installed mysqlclient-1.4.2 2、檢查是否安裝成功
#在虛拟環境中打開python導入對應子產品，看是否成功
(automatic) C:\Users\Administrator>python
Python 3.5.2 (v3.5.2:4def2a2901a5, Jun 25 2016, 22:18:55) [MSC v.1900 64 bit (AM
D64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import MySQLdb #導入成功
>>> 更多參考：https://mysqlclient.readthedocs.io/user_guide.html#installation
二、python操作mysql
class MetaSingleTon(type):
_instance = {}
def __call__(self, *args, **kwargs): # self指代的是傳過來的類 MysqlDataBase
if self not in self._instance:
self._instance[self] = super(
MetaSingleTon, self).__call__(
*args, **kwargs)
return self._instance[self] mysql單例模式
import MySQLdb
class MysqlDataBase(metaclass=MetaSingleTon):
connection = None
def connect(self):
if self.connection is None:
self.connection = MySQLdb.connect(
host="localhost",
port=3306,
user="root",
# password="123456",
db="yw_crm",
charset='utf8')
self.cursor = self.connection.cursor()
return self.cursor
def excute_query_one_data(self):
"""
1、準備sql語句
2、找到cursor
3、執行sql語句
4、拿到結果
5、處理資料
#關閉cursor和conn
:return:擷取單條資料
"""
sql = "select * from crm_userinfo"
self.connect().execute(sql)
# result=self.connect().fetchone()#擷取的是元組資料
result = dict(zip([k[0] for k in self.connect().description],
self.connect().fetchone())) # 将資料處理為字典格式
self.connect().close()
self.close()
return result
def excute_query_more_data(self, page, page_size):
"""
1、準備sql語句
2、找到cursor
3、執行sql語句
4、拿到結果
5、處理資料
#關閉cursor和conn
:return:擷取多條資料
"""
offset = (page - 1) * page_size
sql = 'SELECT * FROM crm_userinfo WHERE id>1 ORDER BY id DESC LIMIT %s, %s;'
cursor = self.connect()
cursor.execute(sql, (offset, page_size))
result = [dict(zip([k[0] for k in cursor.description], row))
for row in cursor.fetchall()] # 将資料處理為字典格式
self.close()
return result
def excute_add_data(self):
# 準備sql
try:
sql = (
"INSERT INTO `crm_userinfo` (`username`,`password`,`email`,`name`,`phone`,`gender`,`department_id`) VALUE "
"( %s, %s, %s, %s, %s,%s,%s );")
# 擷取連接配接和cursor
cursor = self.connect()
# 執行sql
cursor.execute(
sql, ('王豔', '5689', '[email protected]', '王豔', '25412365', 2, 9))
# 送出事務
self.connection.commit()
# 關閉cursor
cursor.close()
except BaseException:
self.connection.rollback() # 上面有事務送出，如果出現異常，所有的全部復原，要麼都成功，要麼都失敗
self.connection.close()
def close(self):
self.connect().close()
self.connection.close()
def main():
db = MysqlDataBase() # 無論建立多少個資料庫執行個體，其結果所有執行個體的位址都相同
# result=db.excute_query_one_data()
# print(result)
# print(result["username"])
###分頁輸出###
# result=db.excute_query_more_data(2,3)
# for item in result:
# print(item)
# 添加資料
# db.excute_add_data()
if __name__ == '__main__':
main() python操作mysql
三、ORM操作
1、sqlalchemy安裝
#進入虛拟環境進行安裝
(automatic) C:\Users\Administrator>pip install SQLALchemy 2、檢測是否安裝成功
#進入python，導入對應的包
(automatic) C:\Users\Administrator>python
Python 3.5.2 (v3.5.2:4def2a2901a5, Jun 25 2016, 22:18:55) [MSC v.1900 64 bit (AM
D64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import sqlalchemy
>>> 3、模型介紹
（1）資料類型
（2）建立模型
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base): #Base是基類
"""
建立資料庫表模型
"""
__tablename__ = 'crm_userinfo' #表示建立的表名
id = Column(Integer, primary_key=True)
username = Column(String(32))
password = Column(String(64))
email = Column(String(32))
name = Column(String(32))
phone = Column(String(32))
gender = Column(Integer)
department_id = Column(Integer) 當然，建立模型表後需要在資料庫中生成表：
engine = create_engine("mysql://root:@127.0.0.1:3306/user_orm?charset=utf8") #連接配接資料庫
Base.metadata.create_all(engine) # 生成對應的表 4、操作模型
在上述模型表的基礎上對表模型進行CURD操作，通過建構一個類進行完成：
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker # 添加資料相關
engine = create_engine("mysql://root:@127.0.0.1:3306/user_orm?charset=utf8") #連接配接資料庫
Session = sessionmaker(bind=engine) #通過session操作模型表，類似mysql中的cursor
class MysqlOrmDb:
"""對模型表的CURD操作"""
def __init__(self):
self.session = Session()
def add_one(self):
"""
新增一條記錄
:return:
"""
new_obj = User(
username="王鵬",
password="21546",
email="[email protected]",
name="王鵬",
phone="12873654",
gender=2,
department_id=3,
)
self.session.add(new_obj)
self.session.commit()
return new_obj
def add_all(self):
user_obj_list = [
User(
username="王漢",
password="21546",
email="[email protected]",
name="王漢",
phone="12873654",
gender=2,
department_id=3,
),
User(
username="李哈",
password="21546",
email="[email protected]",
name="李哈",
phone="12873654",
gender=2,
department_id=3,
)
]
self.session.add_all(user_obj_list)
self.session.commit()
return user_obj_list
def get_one(self):
"""
查詢一條資料
:return:
"""
return self.session.query(User).get(3) # get的參數為id
def get_more(self):
"""
擷取多條資料
:return:
"""
return self.session.query(User).filter_by(gender=2)
def update_one_data(self, pk):
"""
修改一條資料
:param pk:
:return:
"""
new_obj = self.session.query(User).get(pk)
if new_obj:
new_obj.gender = 1
self.session.add(new_obj)
self.session.commit()
return True
return False
def update_more_data(self):
"""
修改多條資料
:return:
"""
data_list = self.session.query(User).filter_by(gender=1)
if data_list:
for item in data_list:
item.gender = 2
self.session.add(item)
self.session.commit()
return True
return False
def delete_one_data(self, pk):
"""
删除一條資料
:param pk:
:return:
"""
user_obj = self.session.query(User).get(pk)
if user_obj:
self.session.delete(user_obj)
self.session.commit()
return True
return False
def delete_more_data(self):
"""
删除多條資料
:return:
"""
data_list = self.session.query(User).filter(User.id > 1)
if data_list:
for item in data_list:
self.session.delete(item)
self.session.commit()
return True
return False 4、完整執行個體
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker # 添加資料相關
engine = create_engine("mysql://root:@127.0.0.1:3306/user_orm?charset=utf8") #連接配接資料庫
Base = declarative_base()
class User(Base): #Base是基類
"""
建立資料庫表
"""
__tablename__ = 'crm_userinfo' #表示建立的表名
id = Column(Integer, primary_key=True)
username = Column(String(32))
password = Column(String(64))
email = Column(String(32))
name = Column(String(32))
phone = Column(String(32))
gender = Column(Integer)
department_id = Column(Integer)
Session = sessionmaker(bind=engine) #通過session操作模型表，類似mysql中的cursor
class MysqlOrmDb:
"""對模型表的CURD操作"""
def __init__(self):
self.session = Session()
def add_one(self):
"""
新增一條記錄
:return:
"""
new_obj = User(
username="王鵬",
password="21546",
email="[email protected]",
name="王鵬",
phone="12873654",
gender=2,
department_id=3,
)
self.session.add(new_obj)
self.session.commit()
return new_obj
def add_all(self):
user_obj_list = [
User(
username="王漢",
password="21546",
email="[email protected]",
name="王漢",
phone="12873654",
gender=2,
department_id=3,
),
User(
username="李哈",
password="21546",
email="[email protected]",
name="李哈",
phone="12873654",
gender=2,
department_id=3,
)
]
self.session.add_all(user_obj_list)
self.session.commit()
return user_obj_list
def get_one(self):
"""
查詢一條資料
:return:
"""
return self.session.query(User).get(3) # get的參數為id
def get_more(self):
"""
擷取多條資料
:return:
"""
return self.session.query(User).filter_by(gender=2)
def update_one_data(self, pk):
"""
修改一條資料
:param pk:
:return:
"""
new_obj = self.session.query(User).get(pk)
if new_obj:
new_obj.gender = 1
self.session.add(new_obj)
self.session.commit()
return True
return False
def update_more_data(self):
"""
修改多條資料
:return:
"""
data_list = self.session.query(User).filter_by(gender=1)
if data_list:
for item in data_list:
item.gender = 2
self.session.add(item)
self.session.commit()
return True
return False
def delete_one_data(self, pk):
"""
删除一條資料
:param pk:
:return:
"""
user_obj = self.session.query(User).get(pk)
if user_obj:
self.session.delete(user_obj)
self.session.commit()
return True
return False
def delete_more_data(self):
"""
删除多條資料
:return:
"""
data_list = self.session.query(User).filter(User.id > 1)
if data_list:
for item in data_list:
self.session.delete(item)
self.session.commit()
return True
return False
def main():
# Base.metadata.create_all(engine) # 生成對應的表
# 添加一條資料
# obj = MysqlOrmDb()
# result = obj.add_one()
# print(result.id)
# 添加多條資料
# obj = MysqlOrmDb()
# result=obj.add_all()
# print(result) #[<__main__.User object at 0x0000000003AE4208>, <__main__.User object at 0x0000000003AE4278>]
# 查詢一條資料
# obj = MysqlOrmDb()
# result=obj.get_one()
# if result:
# print(result.id,result.username)
# else:
# print("not exist")
# 查詢多條資料
# obj=MysqlOrmDb()
# result=obj.get_more()
# print(result.count())
# for user_obj in result:
# print(user_obj.id,user_obj.username,'\n')
obj=MysqlOrmDb()
if __name__ == '__main__':
main() 執行個體
更多請檢視：https://docs.sqlalchemy.org/en/13/intro.html
作者：iveBoy
出處：http://www.cnblogs.com/shenjianping/
本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須在文章頁面給出原文連接配接，否則保留追究法律責任的權利。