python操作mysql数据库
Posted shenjianping
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python操作mysql数据库相关的知识，希望对你有一定的参考价值。
一、环境依赖
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]
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]qq.com‘, ‘王艳‘, ‘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()
三、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、通过orm操作模型表
在上述模型表的基础上对表模型进行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
以上是关于python操作mysql数据库的主要内容，如果未能解决你的问题，请参考以下文章
Python数据库操作 初识mysql和mysql基本操作#学习猿地
python 操作mysql 数据库 安装 MySQL-python