Python使用sqlalchemy模块连接数据库操作示例
本文实例讲述了Python使用sqlalchemy模块连接数据库操作。分享给大家供大家参考,具体如下:
安装:
pipinstallsqlalchemy #安装数据库驱动: pipinstallpymysql pipinstallcx_oracle
举例:(在url后面加入?charset=utf8可以防止乱码)
fromsqlalchemyimportcreate_engine engine=create_engine('mysql+pymysql://username:password@hostname:port/dbname',echo=True)#echo=True打印sql语句信息
create_engine接受一个url,格式为:
#'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名' #常用的 engine=create_engine('sqlite:///:memory:',echo=True)#sqlite内存 engine=create_engine('sqlite:///./cnblogblog.db',echo=True)#sqlite文件 engine=create_engine("mysql+pymysql://username:password@hostname:port/dbname",echo=True)#mysql+pymysql engine=create_engine('mssql+pymssql://username:password@hostname:port/dbname',echo=True)#mssql+pymssql engine=create_engine('postgresql://scott:tiger@hostname:5432/dbname')#postgresql示例 engine=create_engine('oracle://scott:tiger@hostname:1521/sidname')#oracle engine=create_engine('oracle+cx_oracle://scott:tiger@tnsname')#pdb就可以用tns连接
简单demo:
fromsqlalchemyimportcreate_engine,Column,Integer,String fromsqlalchemy.ormimportsessionmaker fromsqlalchemy.ext.declarativeimportdeclarative_base engine=create_engine('oracle://spark:a@orclpdb',echo=True)#echo要求打印sql语句等调试信息 session_maker=sessionmaker(bind=engine) session=session_maker() Base=declarative_base() #对应一张表 classStudent(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''%(self.id,self.name,self.age) Base.metadata.create_all(engine)#若存在STUDENT表则不做,不存在则创建。 queryObject=session.query(Student).order_by(Student.id.desc()) forinsinqueryObject: print(ins.id,ins.name,ins.age) ''' 4hey24 3lwtxxs27 2gyb89 1ns23 '''
将查询结果映射为DataFrame:
importpandasaspd df=pd.read_sql(session.query(Student).filter(Student.id>1).statement,engine) print(df) ''' STUIDSTUNAMESTUAGE 04hey24 12gyb89 23lwtxxs27 '''
查询:
session的query方法除了可以接受Base子类对象作为参数外,还可以是字段,如:
query=session.query(Student.name,Student.age)#query为一个sqlalchemy.orm.query.Query对象 forstu_name,stu_ageinquery: print(stu_name,stu_age)
查询条件filter:
#=/like query.filter(Student.name=='wendy') query.filter(Student.name.like('%ed%')) #in query.filter(Student.name.in_(['wendy','jack'])) query.filter(Student.name.in_( session.query(User.name).filter(User.name.like('%ed%')) )) #notin query.filter(~Student.name.in_(['ed','wendy','jack'])) #isnull/isnotnull query.filter(Student.name==None) query.filter(Student.name.is_(None)) query.filter(Student.name!=None) query.filter(Student.name.isnot(None)) #and fromsqlalchemyimportand_,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) #or query.filter(or_(Student.name=='ed',Student.name=='wendy')) #match query.filter(Student.name.match('wendy'))
Query的方法:
all()方法以列表形式返回结果集:
fromsqlalchemyimportor_,and_ queryObject=session.query(Student).filter(or_(Student.id==1,Student.id==2)) print(queryObject.all())#[, ] queryObject=session.query(Student.name).filter(or_(Student.id==1,Student.id==2)) print(queryObject.all())#[('ns',),('gyb',)]
first()方法返回单个结果。(若结果集为空则返回None)
print(queryObject.first())#('ns',)
one()方法返回单个结果,与first()方法不同的是:当结果集中没有元素或有多于一个元素会抛出异常。
one_or_none()方法同one()一样,不同是结果集为空则返回None,为多个抛出异常。
查询数量:
fromsqlalchemyimportfunc session.query(func.count(Student.id)).scalar()#SELECTcount("STUDENT"."STUID")AScount_1FROM"STUDENT"
分组:
session.query(func.count(Student.id),Student.name).group_by(Student.name).all()
嵌套SQL语句:
fromsqlalchemyimporttext query=session.query(Student.id,Student.name).filter(text('stuid>2')) query=session.query('stuid','stuname','stuage').from_statement(\ text("select*fromstudentwherestuname=:stuname")).params(stuname='hey').all()#[(4,'hey',24)]
更多关于Python相关内容感兴趣的读者可查看本站专题:《Python常见数据库操作技巧汇总》、《Python数学运算技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》及《Python文件与目录操作技巧汇总》
希望本文所述对大家Python程序设计有所帮助。