6.Python操作数据库

1.操作mysql数据库

import pymysql

# 连接数据库
conn=pymysql.connect(host="127.0.0.1",port=3306,user="root",password="123456",database="a",autocommit=True)

# 创建游标进行数据库操作以及获取数据
cursor=conn.cursor(cursor=pymysql.cursors.DictCursor)

# 执行sql语句,并返回影响的行数
# cursor.execute("""insert into user(username,password,email) values(%s,%s,%s)""",("haha","7890","[email protected]",))
# 插入多条数据
# row=cursor.executemany("""insert into user(username,password,email) values(%s,%s,%s)""",[("haha","7890","[email protected]",),("gg","7890","[email protected]",)])

# 查询数据
cursor.execute("select * from user")
# fetchone返回一条数据,类型是字典
result=cursor.fetchone()
print(type(result),result)

# fetchmany返回多条数据,数据展示是列表里面包含字典
result=cursor.fetchmany(3)
print(type(result),result)

# fetchmany返回多条数据,数据展示是列表里面包含字典
result=cursor.fetchall()
print(type(result),result)

# 提交数据
# conn.commit()

# 关闭游标
cursor.close()

# 关闭数据库连接
conn.close()

 2.操作oracle数据库

import cx_Oracle

# 创建数据库连接的类
class connDataBase():
    def __init__(self,userName,password,tns):
        self._userName=userName
        self._password=password
        self._tns=tns
        self.conn,self.cur=self.conn()
     
    #  创建数据库连接
    def conn(self):
        try:
            conn=cx_Oracle.Connection(self._userName,self._password,self._tns)
            cur=conn.cursor()
            return conn,cur
        except Exception as e:
            print("连接数据库出错",e)
            return False
        
    #     增删改数据
    def dealData(self,sql):
        try:
            self.cur.excute(sql)
            self.cur.commit()
        except Exception as e:
            print("处理数据出错")
     
    # 查询数据
    def queryData(self,sql,num="Many"):
        rs=""
        try:
            self.cur.excute(sql)
        except Exception as e:
            print("查询数据出错",e)
            return False
        if num=="Many":
            rs=self.cur.fetchall()
        else:
            rs=self.cur.fetchone()
        return rs
    
    def deleteCur(self):
        self.cur.close()
        self.conn.close()

3.操作cassandra数据库

from cassandra import ConsistencyLevel
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.query import SimpleStatement

def connCassandra(cql):
    # 配置cassandra集群的ip
    contact_points=['ip1','ip2']
    # 配置登陆集群的账号和密码
    auth_provider=PlainTextAuthProvider(username='',password='')
    # 创建一个cassandra的cluster
    cluster=Cluster(contact_points=contact_points,auth_provider=auth_provider)
    # 连接并创建一个会话
    session=cluster.connect()
    # 定义一条cql查询语句
    simple_statement=SimpleStatement(cql,consistency_level=ConsistencyLevel.One)
    # 执行cql语句
    excute_result=session.execute(simple_statement,timeout=None)
    # 获取执行结果
    result=excute_result._current_rows