import timeimport pymssql#import decimalclass 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,'沒有目標(biāo)數(shù)據(jù)庫(kù)')    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,'數(shù)據(jù)庫(kù)訪問失敗')    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('執(zhí)行完成:')

# -*- coding: UTF-8 -*-import cx_Oracle as oracleimport sysimport osimport iosys.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，建立數(shù)據(jù)庫(kù)

在使用的時(shí)候，根據(jù)需求先定義好Model，然后可以通過create_tables()創(chuàng)建表，若是已經(jīng)創(chuàng)建好數(shù)據(jù)庫(kù)表了，可以通過python -m pwiz腳本工具直接創(chuàng)建Model。

第一種方式：

先定義Model，然后通過db.create_tables()創(chuàng)建或Model.create_table()創(chuàng)建表。
例如，我們需要建一個(gè)Person表，里面有name、birthday和is_relative三個(gè)字段，我們定義的Model如下：

然后，我們就可以創(chuàng)建表了

# 創(chuàng)建表Person.create_table()# 創(chuàng)建表也可以這樣, 可以創(chuàng)建多個(gè)# database.create_tables([Person])

其中，CharField、DateField、BooleanField等這些類型與數(shù)據(jù)庫(kù)中的數(shù)據(jù)類型一一對(duì)應(yīng)，我們直接使用它就行，至于CharField => varchar(255)這種轉(zhuǎn)換Peewee已經(jīng)為我們做好了 。

第二種方式：

已經(jīng)存在過數(shù)據(jù)庫(kù)，則直接通過python -m pwiz批量創(chuàng)建Model。
例如，上面我已經(jīng)創(chuàng)建好了test庫(kù)，并且創(chuàng)建了Person表，表中擁有id、name、birthday和is_relative字段。那么，我可以使用下面命令：

然后，輸入密碼，pwiz腳本會(huì)自動(dòng)創(chuàng)建Model，內(nèi)容如下：

from peewee import *database = MySQLDatabase('test', **{'charset': 'utf8', 'use_unicode': True, 'host': 'localhost', 'user': 'root', 'password': ''})class UnknownField(object):    def __init__(self, *_, **__): passclass BaseModel(Model):    class Meta:        database = databaseclass Person(BaseModel):    birthday = DateField()    is_relative = IntegerField()    name = CharField()    class Meta:        table_name = 'person'

操作數(shù)據(jù)庫(kù)

操作數(shù)據(jù)庫(kù)，就是增、刪、改和查。

1、增

直接創(chuàng)建示例，然后使用save()就添加了一條新數(shù)據(jù)

2、刪

使用delete().where().execute()進(jìn)行刪除，where()是條件，execute()負(fù)責(zé)執(zhí)行語(yǔ)句。若是已經(jīng)查詢出來的實(shí)例，則直接使用delete_instance()刪除。

# 刪除姓名為perter的數(shù)據(jù)Person.delete().where(Person.name == 'perter').execute()# 已經(jīng)實(shí)例化的數(shù)據(jù), 使用delete_instancep = Person(name='liuchungui', birthday=date(1990, 12, 20), is_relative=False)p.id = 1p.save()p.delete_instance()

3、改

若是，已經(jīng)添加過數(shù)據(jù)的的實(shí)例或查詢到的數(shù)據(jù)實(shí)例，且表?yè)碛?code>primary key時(shí)，此時(shí)使用save()就是修改數(shù)據(jù)；若是未擁有實(shí)例，則使用update().where()進(jìn)行更新數(shù)據(jù)。

4、查

單條數(shù)據(jù)使用Person.get()就行了，也可以使用Person.select().where().get()。

若是查詢多條數(shù)據(jù)，則使用Person.select().where()，去掉get()就行了。語(yǔ)法很直觀，select()就是查詢，where是條件，get是獲取第一條數(shù)據(jù)。

# 查詢單條數(shù)據(jù)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)# 查詢多條數(shù)據(jù)persons = Person.select().where(Person.is_relative == True)for p in persons:    print(p.name, p.birthday, p.is_relative)

1、連接數(shù)據(jù)庫(kù)在sqlalchemy中，session用于創(chuàng)建程序與數(shù)據(jù)庫(kù)之間的會(huì)話。所有對(duì)象的載入和保存都需要通過session對(duì)象。from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker# 鏈接數(shù)據(jù)庫(kù)采用pymysq模塊做映射，后面參數(shù)是最大連接數(shù)5ENGINE=create_engine('mysql+pymysql://root@127.0.0.1:3306/digchouti?charset=utf8', max_overflow=5)Session = sessionmaker(bind=engine)session = Session()2、創(chuàng)建映射(創(chuàng)建表)一個(gè)映射對(duì)應(yīng)著一個(gè)Python類，用來表示一個(gè)表的結(jié)構(gòu)。下面創(chuàng)建一個(gè)person表，包括id和name兩個(gè)字段。也就是說創(chuàng)建表就是用python的的類來實(shí)現(xiàn)import sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerENGINE=create_engine('mysql+pymysql://root@127.0.0.1:3306/digchouti?charset=utf8', max_overflow=5)#生成一個(gè)SQLORM基類，創(chuàng)建表必須繼承他，別問我啥意思就是這么規(guī)定的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此代碼是創(chuàng)建了一個(gè)名字叫userinfo的表，表里有兩列，一列是id，一列是name。3、添加數(shù)據(jù)當(dāng)然我們創(chuàng)建了表，肯定也要添加數(shù)據(jù),代碼如下：#創(chuàng)建一個(gè)person對(duì)象person = Person(name='張三')#添加person對(duì)象，但是仍然沒有提交到數(shù)據(jù)庫(kù)session.add(person)#提交數(shù)據(jù)庫(kù)session.commit()當(dāng)然還能添加多條數(shù)據(jù)：session.add_all([    Person(name='張三'),    Person(name='aylin')])session.commit()
4、查找數(shù)據(jù)在sqlalchemy模塊中，查找數(shù)據(jù)給提供了query()的方法 下面我就把能用到的給列舉一下：#獲取所有數(shù)據(jù)session.query(Person).all()#獲取name=‘張巖林’的那行數(shù)據(jù)session.query(Person).filter(Person.name=='張三').one()#獲取返回?cái)?shù)據(jù)的第一行session.query(Person).first()#查找id大于1的所有數(shù)據(jù)session.query(Person.name).filter(Person.id>1).all()#limit索引取出第一二行數(shù)據(jù)session.query(Person).all()[1:3]#order by,按照id從大到小排列session.query(Person).ordre_by(Person.id)#equal/like/inquery = 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 orfrom 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計(jì)算個(gè)數(shù)session.query(Person).count()# 修改updatesession.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 funcret = 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()