本文實(shí)例講述了Python使用sqlalchemy模塊連接數(shù)據(jù)庫(kù)操作。分享給大家供大家參考，具體如下：
安裝：
pip install sqlalchemy# 安裝數(shù)據(jù)庫(kù)驅(qū)動(dòng)：pip install pymysqlpip install cx_oracle
舉例：（在url后面加入?charset=utf8可以防止亂碼）
from sqlalchemy import create_engineengine=create_engine('mysql+pymysql://username:password@hostname:port/dbname', echo=True) #echo=True 打印sql語(yǔ)句信息 create_engine接受一個(gè)url，格式為：
# '數(shù)據(jù)庫(kù)類型+數(shù)據(jù)庫(kù)驅(qū)動(dòng)名稱://用戶名:口令@機(jī)器地址:端口號(hào)/數(shù)據(jù)庫(kù)名'# 常用的engine = create_engine('sqlite:///:memory:', echo=True) # sqlite內(nèi)存engine = create_engine('sqlite:///./cnblogblog.db',echo=True) # sqlite文件engine = create_engine("mysql+pymysql://username:password@hostname:port/dbname",echo=True) # mysql+pymysqlengine = create_engine('mssql+pymssql://username:password@hostname:port/dbname',echo=True) # mssql+pymssqlengine = create_engine('postgresql://scott:tiger@hostname:5432/dbname') # postgresql示例engine = create_engine('oracle://scott:tiger@hostname:1521/sidname') # oracleengine = create_engine('oracle+cx_oracle://scott:tiger@tnsname') #pdb就可以用tns連接簡(jiǎn)單demo：
from sqlalchemy import create_engine, Column, Integer, Stringfrom sqlalchemy.orm import sessionmakerfrom sqlalchemy.ext.declarative import declarative_base engine = create_engine('oracle://spark:a@orclpdb',echo=True) #echo要求打印sql語(yǔ)句等調(diào)試信息session_maker = sessionmaker(bind=engine)session = session_maker()Base = declarative_base()#對(duì)應(yīng)一張表class Student(Base): __tablename__ = 'STUDENT' id = Column('STUID', Integer, primary_key=True) name = Column('STUNAME', String(32), nullable=False) age = Column('STUAGE', Integer) def __repr__(self): return '<Student(id:%s, name:%s, age:%s)>' % (self.id, self.name, self.age)Base.metadata.create_all(engine) #若存在STUDENT表則不做，不存在則創(chuàng)建。queryObject = session.query(Student).order_by(Student.id.desc())for ins in queryObject: print(ins.id, ins.name, ins.age)'''4 hey 243 lwtxxs 272 gyb 891 ns 23'''將查詢結(jié)果映射為DataFrame：
import pandas as pddf = pd.read_sql(session.query(Student).filter(Student.id > 1).statement, engine) print(df)''' STUID STUNAME STUAGE0 4 hey 241 2 gyb 892 3 lwtxxs 27'''
查詢：
session的query方法除了可以接受Base子類對(duì)象作為參數(shù)外，還可以是字段，如：
query = session.query(Student.name, Student.age) # query為一個(gè)sqlalchemy.orm.query.Query對(duì)象for stu_name, stu_age in query: print(stu_name, stu_age)
查詢條件filter：
# = / likequery.filter(Student.name == 'wendy')query.filter(Student.name.like('%ed%'))# inquery.filter(Student.name.in_(['wendy', 'jack']))query.filter(Student.name.in_( session.query(User.name).filter(User.name.like('%ed%'))))# not inquery.filter(~Student.name.in_(['ed', 'wendy', 'jack']))# is null / is not nullquery.filter(Student.name == None)query.filter(Student.name.is_(None))query.filter(Student.name != None)query.filter(Student.name.isnot(None))# andfrom sqlalchemy import and_, or_query.filter(and_(Student.name == 'ed', Student.age != 23))query.filter(Student.name == 'ed', Student.age != 23)query.filter(Student.name == 'ed').filter(Student.age != 23)# orquery.filter(or_(Student.name == 'ed', Student.name == 'wendy'))# matchquery.filter(Student.name.match('wendy'))Query的方法：
all()方法以列表形式返回結(jié)果集：
from sqlalchemy import or_, and_queryObject = session.query(Student).filter(or_(Student.id == 1, Student.id == 2))print(queryObject.all()) # [<Student(id:1, name:ns, age:23)>, <Student(id:2, name:gyb, age:89)>]queryObject = session.query(Student.name).filter(or_(Student.id == 1, Student.id == 2))print(queryObject.all()) # [('ns',), ('gyb',)] first()方法返回單個(gè)結(jié)果。（若結(jié)果集為空則返回None）
print(queryObject.first()) # ('ns',) one()方法返回單個(gè)結(jié)果，與first()方法不同的是：當(dāng)結(jié)果集中沒(méi)有元素或有多于一個(gè)元素會(huì)拋出異常。
one_or_none()方法同one()一樣，不同是結(jié)果集為空則返回None，為多個(gè)拋出異常。
查詢數(shù)量：
from sqlalchemy import funcsession.query(func.count(Student.id)).scalar() # SELECT count("STUDENT"."STUID") AS count_1 FROM "STUDENT"分組：
session.query(func.count(Student.id), Student.name).group_by(Student.name).all()
嵌套SQL語(yǔ)句：
from sqlalchemy import textquery = session.query(Student.id, Student.name).filter(text('stuid>2'))query = session.query('stuid', 'stuname', 'stuage').from_statement(/text("select * from student where stuname=:stuname")).params(stuname='hey').all() #[(4, 'hey', 24)]希望本文所述對(duì)大家Python程序設(shè)計(jì)有所幫助。
新聞熱點(diǎn)
疑難解答
圖片精選
