Python3 连接各类数据库

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# 安装PyMySQL
# sudo pip install PyMySQL

import pymysql


config = {

'host':'localhost',
'port':3306,
'user':'root',
'passwd':'root',
'db':'Geek_Web',
'charset':'utf8mb4',
# 数据库内容以字典格式输出
#'cursorclass':pymysql.cursors.DictCursor,
}

# 连接数据库
def Mysql():
# 连接数据库
#db = pymysql.connect(host="localhost", port=3306, user="root", passwd="root", db="Geek_Web", charset="utf8mb4")
db = pymysql.connect(**config)
#cursor()方法获取操作游标 
cursor = db.cursor()

try:

return (db, cursor)

except:

print("数据库访问失败")

# 增
def Insert(db, cursor):
sql = "insert into main(id, Tag, Name, Version, Introduce, Class, Worked_OS, Course_URL, Download_URL, Image_URL, Remarks_1, Remarks_2) 
values (NULL, '软件编号', '软件名称', '软件版本', '软件简介', '软件类别', '运行环境', '教程地址', '下载地址', '图标地址', '备注1', '备注2')"
# 执行SQL语句
cursor.execute(sql)
# 没有设置默认自动提交，需要主动提交，以保存所执行的语句
db.commit()

# 删
def Delect(db, cursor):
sql = "DELETE FROM main WHERE Name = '修改后的名字'"
cursor.execute(sql)
db.commit()

# 查
def Select(db, cursor):
sql = "SELECT * FROM main"
cursor.execute(sql)
# 获取所有记录列表
results = cursor.fetchall()

return results

# 改
def Update(db, cursor):
sql = "UPDATE main SET Name = '修改后的名字' WHERE Remarks_2 = '备注2'"
cursor.execute(sql)
db.commit()

# 关闭数据库连接
def Close(db, cursor):
cursor.close()
db.close()

(db, cursor) = Mysql()
print("
-------------数据库初始状态-------------")
print(Select(db, cursor))
Insert(db, cursor)
print("
-------------数据库插入数据-------------")
print(Select(db, cursor))
Update(db, cursor)
print("
-------------数据库修改数据-------------")
print(Select(db, cursor))
Delect(db, cursor)
print("
-------------数据库删除数据-------------")
print(Select(db, cursor))
Close(db, cursor)

import time
import pymssql
#import decimal
class MSSQL:
  def __init__(self,host,user,pwd,db):
    self.host=host
    self.user=user
    self.pwd=pwd
    self.db=db
  def GetConnect(self):
    if not self.db:
      raise(NameError,'没有目标数据库')
    self.connect=pymssql.connect(host=self.host,user=self.user,password=self.pwd,database=self.db,charset='utf8')
    cur=self.connect.cursor()
    if not cur:
      raise(NameError,'数据库访问失败')
    else:
      return cur
  def ExecSql(self,sql):
     cur=self.GetConnect()
     cur.execute(sql)
     self.connect.commit()
     self.connect.close()
  def ExecQuery(self,sql):
    cur=self.GetConnect()
    cur.execute(sql)
    resList = cur.fetchall()
    self.connect.close()
    return resList  
def main():
  ms = MSSQL(host="192.168.0.108", user="sa", pwd="sa", db="ComPrject")
  resList = ms.ExecQuery("select *from TestModel")
  print(resList)
if __name__ == '__main__':
  main()
  input("执行完成:")

import cx_Oracle
#连接数据库，下面括号里内容根据自己实际情况填写
conn = cx_Oracle.connect('用户名/密码@IP:端口号/SERVICE_NAME')
# 使用cursor()方法获取操作游标
cursor = conn.cursor()
#使用execute方法执行SQL语句
result=cursor.execute('Select member_id from member')
#使用fetchone()方法获取一条数据
#data=cursor.fetchone()
 
#获取所有数据
all_data=cursor.fetchall()
 
#获取部分数据，8条
#many_data=cursor.fetchmany(8)

print (all_data)
db.close()

# -*- coding: UTF-8 -*-
import cx_Oracle as oracle
import sys
import os
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
class ExportOracle:
    def __init__(self,odbc,user):
        self.odbc = odbc
        self.user = user
 
    def start(self):
        db = oracle.connect(self.odbc)
        cursor = db.cursor()
        cursor.execute("SELECT NAME,AGE FROM USERS ")
        datas = cursor.fetchall()
        print(datas)
 
if __name__ == '__main__':
    orcleDb_config = {
        'odbc':'TEST_XIAOYI/TEST_XIAOYI@192.168.1.1:1521/orcl',
        'user': 'TEST_XIAOYI',
    }
    mtables = ExportOracle(orcleDb_config['odbc'],orcleDb_config['user']);
    mtables.start();

定义Model，建立数据库

在使用的时候，根据需求先定义好Model，然后可以通过create_tables()创建表，若是已经创建好数据库表了，可以通过python -m pwiz脚本工具直接创建Model。

第一种方式：

先定义Model，然后通过db.create_tables()创建或Model.create_table()创建表。
例如，我们需要建一个Person表，里面有name、birthday和is_relative三个字段，我们定义的Model如下：

from peewee import *

# 连接数据库
database = MySQLDatabase('test', user='root', host='localhost', port=3306)

# 定义Person
class Person(Model):
    name = CharField()
    birthday = DateField()
    is_relative = BooleanField()

    class Meta:
        database = database

然后，我们就可以创建表了

# 创建表
Person.create_table()

# 创建表也可以这样, 可以创建多个
# database.create_tables([Person])

其中，CharField、DateField、BooleanField等这些类型与数据库中的数据类型一一对应，我们直接使用它就行，至于CharField => varchar(255)这种转换Peewee已经为我们做好了 。

第二种方式：

已经存在过数据库，则直接通过python -m pwiz批量创建Model。
例如，上面我已经创建好了test库，并且创建了Person表，表中拥有id、name、birthday和is_relative字段。那么，我可以使用下面命令：

# 指定mysql，用户为root，host为localhost，数据库为test
python -m pwiz -e mysql -u root -H localhost --password test > testModel.py

然后，输入密码，pwiz脚本会自动创建Model，内容如下：

from peewee import *

database = MySQLDatabase('test', **{'charset': 'utf8', 'use_unicode': True, 'host': 'localhost', 'user': 'root', 'password': ''})

class UnknownField(object):
    def __init__(self, *_, **__): pass

class BaseModel(Model):
    class Meta:
        database = database

class Person(BaseModel):
    birthday = DateField()
    is_relative = IntegerField()
    name = CharField()

    class Meta:
        table_name = 'person'

操作数据库

操作数据库，就是增、删、改和查。

1、增

直接创建示例，然后使用save()就添加了一条新数据

# 添加一条数据
p = Person(name='liuchungui', birthday=date(1990, 12, 20), is_relative=True)
p.save()

2、删

使用delete().where().execute()进行删除，where()是条件，execute()负责执行语句。若是已经查询出来的实例，则直接使用delete_instance()删除。

# 删除姓名为perter的数据
Person.delete().where(Person.name == 'perter').execute()

# 已经实例化的数据, 使用delete_instance
p = Person(name='liuchungui', birthday=date(1990, 12, 20), is_relative=False)
p.id = 1
p.save()
p.delete_instance()

3、改

若是，已经添加过数据的的实例或查询到的数据实例，且表拥有primary key时，此时使用save()就是修改数据；若是未拥有实例，则使用update().where()进行更新数据。

# 已经实例化的数据,指定了id这个primary key,则此时保存就是更新数据
p = Person(name='liuchungui', birthday=date(1990, 12, 20), is_relative=False)
p.id = 1
p.save()

# 更新birthday数据
q = Person.update({Person.birthday: date(1983, 12, 21)}).where(Person.name == 'liuchungui')
q.execute()

4、查

单条数据使用Person.get()就行了，也可以使用Person.select().where().get()。

若是查询多条数据，则使用Person.select().where()，去掉get()就行了。语法很直观，select()就是查询，where是条件，get是获取第一条数据。

# 查询单条数据
p = Person.get(Person.name == 'liuchungui')
print(p.name, p.birthday, p.is_relative)

# 使用where().get()查询
p = Person.select().where(Person.name == 'liuchungui').get()
print(p.name, p.birthday, p.is_relative)

# 查询多条数据
persons = Person.select().where(Person.is_relative == True)
for p in persons:
    print(p.name, p.birthday, p.is_relative)

from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
 
Base = declarative_base()
 
class User(Base):
    # 表的名字
    __tablename__ = 'user'
 
    # 表的结构
    id = Column(String(20), primary_key=True)
    name = Column(String(20))
 
# 初始化数据库连接
engine = create_engine('mysql+mysqlconnector://root:111111@localhost:3306/test')
# 创建DBSession类型
DBSession = sessionmaker(bind=engine)
# create_engine用来初始化数据库连接.
# SQLAlchemy用一个字符串表示连接信息'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
 
# # 创建session对象:
# session = DBSession()
# # 创建新User对象
# new_user = User(id='5', name='Bob')
# # 添加到session
# session.add(new_user)
# # 提交保存到数据库
# session.commit()
# # 关闭session
# session.close()
 
# 可见将关键是获取session, 然后把对象添加到session, 最后提交并关闭.(DBSession对象, 可以看做是当前数据库的连接)
 
# 查询
session = DBSession()
# 创建Query查询, filter是where条件, 最后调用one()返回唯一行, 如果调用all()则返回所有行.
user = session.query(User).filter(User.id=='5').one()
print('type:', type(user))
print('name:', user.name)
session.close()
 
# ORM就是把数据库表的行与相应的对象简历关联, 互相转换.
# 由于关系数据库的多个表还可以用外键实现一对多, 多对多的关联, 相应地, ORM框架也可以提供两个对象之间的一对多, 多对多功能.

1、连接数据库
在sqlalchemy中，session用于创建程序与数据库之间的会话。所有对象的载入和保存都需要通过session对象。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 链接数据库采用pymysq模块做映射，后面参数是最大连接数5
ENGINE=create_engine("mysql+pymysql://root@127.0.0.1:3306/digchouti?charset=utf8", max_overflow=5)
Session = sessionmaker(bind=engine)

session = Session()

2、创建映射(创建表)
一个映射对应着一个Python类，用来表示一个表的结构。下面创建一个person表，包括id和name两个字段。也就是说创建表就是用python的的类来实现
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker

ENGINE=create_engine("mysql+pymysql://root@127.0.0.1:3306/digchouti?charset=utf8", max_overflow=5)

#生成一个SQLORM基类，创建表必须继承他，别问我啥意思就是这么规定的
Base = declarative_base()

class Person(Base):
    __tablename__ = 'userinfo'

    id   = Column(Integer, primary_key=True)
    name = Column(String(32))

    def __repr__(self):
        return "<Person(name='%s')>" % self.name
此代码是创建了一个名字叫userinfo的表，表里有两列，一列是id，一列是name。

3、添加数据
当然我们创建了表，肯定也要添加数据,代码如下：
#创建一个person对象
person = Person(name='张三')
#添加person对象，但是仍然没有提交到数据库
session.add(person)
#提交数据库
session.commit()

当然还能添加多条数据：
session.add_all([
    Person(name='张三'),
    Person(name='aylin')
])
session.commit()

4、查找数据
在sqlalchemy模块中，查找数据给提供了query()的方法 下面我就把能用到的给列举一下：
#获取所有数据
session.query(Person).all()

#获取name=‘张岩林’的那行数据
session.query(Person).filter(Person.name=='张三').one()

#获取返回数据的第一行
session.query(Person).first()

#查找id大于1的所有数据
session.query(Person.name).filter(Person.id>1).all()

#limit索引取出第一二行数据
session.query(Person).all()[1:3]

#order by,按照id从大到小排列
session.query(Person).ordre_by(Person.id)

#equal/like/in
query = session.query(Person)
query.filter(Person.id==1).all()
query.filter(Person.id!=1).all()
query.filter(Person.name.like('%ay%')).all()
query.filter(Person.id.in_([1,2,3])).all()
query.filter(~Person.id.in_([1,2,3])).all()
query.filter(Person.name==None).all()

#and or
from sqlalchemy import and_
from sqlalchemy import or_
query.filter(and_(Person.id==1, Person.name=='张三')).all()
query.filter(Person.id==1, Person.name=='张三').all()
query.filter(Person.id==1).filter(Person.name=='张三').all()
query.filter(or_(Person.id==1, Person.id==2)).all()

# count计算个数
session.query(Person).count()

# 修改update
session.query(Person).filter(id > 2).update({'name' : '张三'})

# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()

# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()