一、环境依赖
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/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须在文章页面给出原文连接，否则保留追究法律责任的权利。