python3 数据库的常用操作

一、Mysql数据库：

安装pymysql: 

pip install pymysql

1、数据库连接对象 connect 的常用方法：

cursor() # 创建一个游标对象

commit() # 提交事务

rollback() # 事务回滚

close() # 关闭数据库连接

2、游标对象 cursor 的常用方法：

execute()  # 执行SQL语句

executemany()  # 用来执行多天SQL语句

close() # 用来关闭游标

fetchone() # 用来从结果取出一条记录，并将游标指向下一条记录

fetchall()  # 从结果中取出所有记录

scroll()  # 用于游标滚动

示例如下：

from myconn import myconn
 
class Sql:
	def __init__(self,db=None):
		# 初始化连接数据库
		# self.conn = pymysql.connect(db=dataName, user="user", passwd="passwd", host="host",charset='utf8')
		# 这是我自己写的方法，平常的话，安装pymysql，用上面一句连接就可
		self.conn = myconn.get_conn(db) if db!=None else myconn.get_conn()
 
	def get_databases(self):
		'''  获取所有数据库名 '''
		return self.getSqlData("show databases")
	def get_tables(self):
		'''  获取所有表名 '''
		return self.getSqlData("show tables")
 
	def create_table(self,table):
		''' 创建表 '''
		cur = self.conn.cursor()
		# id：无符号整数，主键，自增；name：字符串；age：无符号整数
		sql = """create table {} (id int unsigned primary key auto_increment name varchar(10),age int unsigned)""".format(table)
		cur.execute(sql)
		self.conn.commit()
 
	def insert_data(self,table):
		''' 增加一条数据到数据表 '''
		cur = self.conn.cursor()
		sql = "insert into {}(name,age) values(%s,%s)"
		cur.execute(sql,('myName',80))
		self.conn.commit()
 
	def update_data(self,table):
		''' 修改表中的数据 '''
		cur = self.conn.cursor()
		sql = "update {} set age=18 where name='myName'".format(table)
		cur.execute(sql)
		self.conn.commit()
 
	def select_data(self, table):
		'''  从数据库表查询数据 '''
		cur = self.conn.cursor()
		sql = "select name,age from {}".format(table)
		cur.execute(sql)
		return cur.fetchall()
 
	def delete_data(self, table):
		'''  从数据库表删除数据 '''
		cur = self.conn.cursor()
		sql = "delete from {} where name='myName'".format(table)
		cur.execute(sql)
		self.conn.commit()
 
 
	def get_fields(self,table):
		'''  获取指定表的字段 '''
		cur = self.conn.cursor()
		sql = "SELECT * FROM {} LIMIT 1".format(table)
		cur.execute(sql)
		v = cur.description
		zds = [i[0] for i in v]
		self.conn.commit()
		return zds
 
 
	def unique(self,table,*fields):
		''' 唯一设置 
		table：表名，fields：字段名列表； '''
		cur = self.conn.cursor()
		if len(fields) == 1:
			sql = "ALTER TABLE {} ADD unique(".format(table)
		else:
			sql = "ALTER TABLE {} ADD UNIQUE KEY(".format(table)
		for i in fields:
			sql += i
			if i != fields[-1]:
				sql += ','
			else:
				sql += ')'
		try:
			cur.execute(sql)
		except Exception as exc:
			print(exc)
		else:
			self.conn.commit()
		
	def closeSql(self):
		''' 关闭数据库连接 '''
		self.conn.close()

二、MongoDB 数据库

1，安装 mongodb：

到MongoDB官网下载对应版本的安装包: https://www.mongodb.com/download-center?jmp=nav#community

1，把MongoDB安装到C盘或者D盘都可以

2，在C盘建立data\db文件夹作为数据文件的存储路径，建立data\log文件夹存储日志文件。

3，安装服务：cmd 进入到MongoDB的bin目录，执行：mongod --dbpath "C:\data\db"  --logpath "C:\data\log\log.txt"  --install -serviceName "MongoDB"

4，开启服务：net start MongoDB

注：有时由于没有正常关闭MongoDB，导致开启服务失败，可尝试删除C:\data\db下的mongod.lock再开启服务

mongod --dbpath "C:\data\db"  --logpath "C:\data\log\log.txt"  --install -serviceName "MongoDB"  

2，安装 pymongo:

pip install pymongo

示例如下：

import pymongo
import datetime
 
class Mongodb:
	""" Python MangoDB 的简单操作 """
 
	def __init__(self):
		# 1、建立连接
		# client = pymongo.MongoClient('localhost',27017) # 第一种方法
		client = pymongo.MongoClient('mongodb://localhost:27017') # 第二种方法
 
		# 2、获取数据库
		# db = client.db_name # 第一种方法
		db = client['db_name'] # 第二种方法
 
		# 3、获取一个集合
		# self.collection = db.table # 第一种方法
		self.collection = db['table'] # 第二种方法
 
	def insert_data(self):
		''' 插入文档 '''
		# name：姓名；age：年龄；datetime：存储时间
		user = {
			"name":"myName",
			"age":18,
			"datetime":datetime.datetime.utcnow()
		}
		# 插入后，如果文档内没有_id这个键值，则系统会自动添加_id；其中user可以是多个user组成的列表
		user_id = self.collection.insert(user)
		return user_id
 
	def find_one_data(self):
		''' 查询一条文档 '''
		data = self.collection.find_one({"name":"myName"}) # 查询name为myName的
		return data
 
	def find_data(self):
		''' 查询多个文档 '''
		# data = [d for d in self.collection.find()] # 查询所有
		data = [d for d in self.collection.find({"name":"myName"})] # 查询所有指定name的文档
		return data
 
	def find_limit_data(self):
		''' 根据条件查询数据：
			MongoDB中条件操作符有：
			(>) 	  大于 - $gt
			(<) 	  小于 - $lt
			(>=)  大于等于 - $gte
			(<= ) 小于等于 - $lte '''
		data = self.collection.find({"age":{"$gt": 12}}) # 查询年龄大于12岁的
		return data
 
	def get_size(self):
		''' 查询符合条件的文档条数 '''
		size = self.collection.find({"name":"myName"}).count()
		return size
 
	def get_names(self):
		''' 查询所有 name 的值，不重复的。返回list '''
		names = self.collection.distinct('name')
		return names
 
	def update_data(self):
		''' 修改文档 '''
		self.collection.update({"name":"myName"},{"$set":{"age":28}}) # 修改name为myName的年龄
 
	def delete_data(self):
		''' 删除文档 '''
		self.collection.remove({"name":"myName"}) # 删除name为myName的文档

三、SQLite 数据库

import sqlite3
 
class Sqlite:
	""" Sqlite 数据库的简单使用 """
 
	def __init__(self):
		# 1、打开 mydb.db 数据库，若不存在则创建
		self.conn = sqlite3.connect('D:\\mydb.db')
		# self.conn = sqlite3.connect(':memory:') # 在内存中创建数据库文件
		
	
	def create_table(self):
		''' 建表 '''
		# 创建游标对象
		cur = self.conn.cursor()
		sql = "create table tb_name (id integer primary key,name varchar(10),age integer)"
		cur.execute(sql)  	# 执行
		self.conn.commit() 	# 提交
 
	def insert_data(self):
		''' 插入数据 '''
		cur = self.conn.cursor()
		sql = "insert into tb_name values(?,?,?)"
		cur.execute(sql,(1,'myName',20))  	# 执行
		self.conn.commit() 					# 提交
 
	def select_data(self):
		''' 查询数据 '''
		cur = self.conn.cursor()
		sql = "select * from tb_name"
		cur.execute(sql)  	# 执行
		data = cur.fetchall() # 返回所有
		return data
 
	def update_data(self):
		''' 修改数据 '''
		cur = self.conn.cursor()
		sql = "update tb_name set name=? where id=?"
		cur.execute(sql,('isName',1))  	# 执行
		self.conn.commit()
 
	def delete_data(self):
		''' 删除数据 '''
		cur = self.conn.cursor()
		sql = "delete from tb_name where id=?"
		cur.execute(sql,(1,))  	# 执行
		self.conn.commit()
 
	def close_db(self):
		''' 关闭数据库连接 '''
		self.conn.close()

四、Redis 数据库

#在Redis中设置值，默认不存在则创建，存在则修改
安装 redis：pip install redis
import redis
1、存入数据
r = redis.Redis(host='localhost',port=6379,db=0)
r.set('name', 'zhangsan')
'''参数：
     set(name, value, ex=None, px=None, nx=False, xx=False)
     ex，过期时间（秒）
     px，过期时间（毫秒）
     nx，如果设置为True，则只有name不存在时，当前set操作才执行,同setnx(name, value)
     xx，如果设置为True，则只有name存在时，当前set操作才执行'''
setex(name, value, time)
#设置过期时间（秒）

psetex(name, time_ms, value)
#设置过期时间（豪秒）
mset()

#批量设置值
r.mset(name1='zhangsan', name2='lisi')
#或
r.mget({"name1":'zhangsan', "name2":'lisi'})
get(name)

2、获取数据

mget(keys, *args)
#批量获取
print(r.mget("name1","name2"))
#或
li=["name1","name2"]
print(r.mget(li))

getset(name, value)
#设置新值，打印原值
print(r.getset("name1","wangwu")) #输出:zhangsan
print(r.get("name1")) #输出:wangwu

getrange(key, start, end)
#根据字节获取子序列
r.set("name","zhangsan")
print(r.getrange("name",0,3))#输出:zhan

setrange(name, offset, value)
#修改字符串内容，从指定字符串索引开始向后替换，如果新值太长时，则向后添加
r.set("name","zhangsan")
r.setrange("name",1,"z")
print(r.get("name")) #输出:zzangsan
r.setrange("name",6,"xxx")

print(r.get("name")) #输出:zzangsxxx

import redis
 
class MyRedis:
	""" Redis 数据库简单使用 """
	def __init__(self):
	        # 1、建立 Redis连接
		# 第一种方式
		# self.red = redis.Redis(host='localhost',port=6379,db=0)
		# 第二种方式(以连接池的方式)
		pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
                self.red = redis.Redis(connection_pool=pool)  
 
	def insert_data(self):
		''' 插入数据 '''
		self.red.set("key","value")
 
	def get_data(self):
		''' 查询数据 '''
		data = self.red.get("key")
		return data
 
	def delete_data(self):
		''' 删除数据 '''
		self.red.delete('key')

Redis 之 消息推送

1、创建服务端用以发布数据

import redis
import time
from datetime import datetime
 
def resis_server(host,port):
	pool = redis.ConnectionPool(host=host,port=port,db=1)
	r = redis.StrictRedis(connection_pool=pool)
	count = 1
	while True:
		dt = datetime.now()
		print(dt) # 要发布的数据
		# 发布到指定频道
		r.publish('channel_name',dt)
		if count >= 10:
			# 只是为了测试，所以限制为10次后，发布结束通知
			r.publish('channel_name','end') 
			print('停止发布')
			break
		time.sleep(1)
		count += 1
		
 
if __name__ == '__main__':
	resis_server('192.168.88.64',6379)

2、创建客户端以接收数据

import redis
 
def redis_client(host,port):
	pool = redis.ConnectionPool(host=host,port=port,db=1)
	r = redis.StrictRedis(connection_pool=pool)
	p = r.pubsub()
	p.subscribe('channel_name') # 订阅频道
	for item in p.listen():
		print(item)  # 输出字典
		if item['type'] == 'message':
			data = item['data']
			print(data)  # 输出发送过来的消息
			if item['data'] == b'end':
				break
	p.unsubscribe('channel_name')
	print('取消订阅')
 
if __name__ == '__main__':
	redis_client('192.168.88.64',6379)

3、分别运行服务端与客户端，最终执行结果如下：（左边：服务端发布的消息，右边：客户的接收的消息）

五、使用 hdf5 快速存取数据

安装 h5py：pip install h5py

# w  以写入方式打开文件
# r  只读，文件必须已存在
# r+ 读写，文件必须已存在
# w  新建文件，若存在覆盖
# w- 或x，新建文件，若存在报错

# a  如存在则读写，不存在则创建(默认)

import h5py
 
def write_h5():
	''' 写入数据，写入一个list '''
	h5 = h5py.File('test.hdf5','w')
	h5['/test/t1'] = [1,2,3,4,5]
	h5.close()
 
def read_h5():
	''' 读取数据 '''
	h5 = h5py.File('test.hdf5','r')
	v = h5['/test/t1'].value
	h5.close()
	return v
 
if __name__ == '__main__':
	write_h5()
	v = read_h5()
	print(v)

执行结果：

[1 2 3 4 5]

转自：

python3 数据库的常用操作

注意：本文归作者所有，未经作者允许，不得转载