本文共 14770 字,大约阅读时间需要 49 分钟。
这篇文章主要介绍了Python的SQLAlchemy框架使用入门,SQLAlchemy框架是Python中用来操作数据库的ORM框架之一,学习过程中主要参考网上现有资料,整理成笔记以便后续自己查阅。
如果转载,请保留作者信息。 邮箱地址:@gmail.com SQLAlchemy: 中文参考:SQLAlchemy的是Python的SQL工具包和对象关系映射,给应用程序开发者提供SQL的强大功能和灵活性。它提供了一套完整的企业级的持久性模式,专为高效率和高性能的数据库访问,改编成简单的Python的领域语言。
SQLAlchemy是Python界的ORM(Object Relational Mapper)框架,它两个主要的组件: SQLAlchemy ORM 和 SQLAlchemy Core 。pip install SQLAlchemy检查安装是否成功: >>> import sqlalchemy>>> sqlalchemy.__version__>>> '1.0.12'
连接MySQL数据库(需要MySQLdb支持):
from sqlalchemy import create_engineDB_CONNECT_STRING = 'mysql+mysqldb://root:password@localhost/test?charset=utf8'engine = create_engine(DB_CONNECT_STRING,echo=True)
create_engine方法返回一个Engine实例,Engine实例只有直到触发数据库事件时才真正去连接数据库。
echo参数是设置 SQLAlchemy 日志记录,这通过 Python 的标准logging模块的快捷方式。启用它,我们会看到产生的所有生成的 SQL,sqlalchemy与数据库通信的命令都将打印出来,例如执行:engine.execute("select 1").scalar()
执行打印信息:
2016-02-28 23:55:37,544 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'2016-02-28 23:55:37,544 INFO sqlalchemy.engine.base.Engine ()2016-02-28 23:55:37,545 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()2016-02-28 23:55:37,546 INFO sqlalchemy.engine.base.Engine ()......
declarative_base类维持了一个从类到表的关系,通常一个应用使用一个base实例,所有实体类都应该继承此类对象
from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()
定义映射的表,表名称、 数据类型的列,在这里定义一个User类:
from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy import create_engineDB_CONNECT_STRING = 'mysql+mysqldb://root:password@localhost/test?charset=utf8'engine = create_engine(DB_CONNECT_STRING,echo=True)Base = declarative_base()class User(Base): __tablename__ = 'users' id = Column(Integer,primary_key=True) name = Column(String(10)) fullname = Column(String(20)) password = Column(String(20)) #可以设定长度 def __init__(self,name,fullname,password): self.name = name self.fullname = fullname self.password = password def __repr(self): return ""%(self.name,self.fullname,self.password)Base.metadata.create_all(engine)
类使用声明式至少需要一个__tablename__属性定义数据库表名字,并至少一Column是主键。User类定义一个__repr__()方法,但请注意,是可选;
sqlalchemy 就是把Base子类转变为数据库表,定义好User类后,会生成Table和mapper(),分别通过User.table 和User.__mapper__返回这两个对象,对于主键,象oracle没有自增长的主键时,要使用:from sqlalchemy import SequenceColumn(Integer,Sequence('user_idseq'),prmary_key=True)
数据表字段长度定义:
Column(String(50))
Base.metadata返回sqlalchemy.schema.MetaData对象,它是所有Table对象的集合,调用create_all()该对象会触发CREATE TABLE语句,如果数据库还不存在这些表的话。
>>> Base.metadata.create_all(engine)>>> CREATE TABLE users ( id INTEGER NOT NULL AUTO_INCREMENT, name VARCHAR(10), fullname VARCHAR(20), password VARCHAR(20), PRIMARY KEY (id))2016-02-29 22:52:03,260 INFO sqlalchemy.engine.base.Engine ()2016-02-29 22:52:03,350 INFO sqlalchemy.engine.base.Engine COMMIT
创建一个User类:
ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')print ed_user.nameprint ed_user.passwordprint str(ed_user.id)输出:>>>ed>>>edspassword>>>None
Session 使用 connection发送query,把返回的result row 填充到一个object中,该对象同时还会保存在Session中,Session内部有一个叫 Identity Map的数据结构,为每一个对象维持了唯一的副本。primary key 作为 key ,value就是该object。session刚开始无状态,直到有query发起时。对象的变化会被session的跟踪维持着,在数据库做下一次查询后者当前的事务已经提交了时,it fushed all pendings changes to the database.
这就是传说中的 Unit of work 模式 例如:def unit_of_work(): session = Session() album = session.query(Album).get(4) album.name = "jun" #这里不会修改album的name属性,不会触发update语句def unit_of_work(): session = Session() album = session.query(Album).get(4) album.name = "jun" #这里修改了album的name属性,会触发一个update语句 session.query(Artist).get(11) session.commit()
规则:始终保持session与function和objecct分离
⚠注意:对象实例有四种状态,分别是:
Transient(瞬时的):实例还不在session中,还没有保存到数据库中去,没有数据库身份,像刚创建出来的对象比如User(),仅仅只有mapper()与之关联. Pending(挂起的):调用session.add()后,Transient对象就会变成Pending,这个时候它还是不会保存到数据库中,只有等到触发了flush动作才会存在数据库,比如query操作就可以出发flush。同样这个时候的实例的主键一样为None Persistent(持久的):session中,数据库中都有对应的一条记录存在,主键有值了。 Detached(游离的):数据库中有记录,但是session中不存在,对这个状态的对象进行操作时,不会触发任何SQL语句。commit()方法调用的时候
查询时会清理缓存,保证查询结果能反映对象的最新状态 显示调用session的flush方法 Session是真正与数据库通信的handler,你还可以把他理解一个容器,add就是往容器中添加对象 执行完add方法后,ed_user对象处于pending状态,不会触发INSERT语句,当然ed_uesr.id也为None,如果在add方后有查询(session.query),那么会flush一下,把数据刷一遍,把所有的pending信息先flush再执行query。from sqlalchemy.orm import sessionmakerSession = sessionmaker(bind=engine)
这个定制的Session类将创建新的Session对象绑定到我们的数据库。
无论何时你需要与数据库连接,你实例化一个Session:session = Session()
上述Session是与我们Mysql连接启用的Engine,但它还没有打开任何连接。它第一次使用时,由维护的连接池检索连接,并获取不放,直到我们提交所有更改和/或关闭会话对象才会释放掉。
添加User对象到我们定义的session上,
ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')session.add(ed_user)
添加多个的User对象,使用add_all() :
session.add_all([User(name='wendy', fullname='Wendy Williams', password='foobar'),User(name='mary', fullname='Mary Contrary', password='xxg527'),User(name='fred', fullname='Fred Flinstone', password='blah')])
session.commit()
commit()刷新任何剩余的变化保持到数据库中,并将提交的事务。
由于Session工作在一个事务内,我们可以回滚所做过的更改。
session.rollback()
在Session上使用query()方法,创建了一个Query对象。此函数接受数目可变的参数,可以是任意组合的类和类表的描述符
Query对象通过Session.query获取,query接收类或属性参数,以及多个类>>> for instance in session.query(User).order_by(User.id):... print(instance.name, instance.fullname)ed Ed Joneswendy Wendy Williamsmary Mary Contraryfred Fred FlinstoneQuery还接受 ORM 检测描述符作为参数,参数返回的结果被表示为元组:>>> for name, fullname in session.query(User.name, User.fullname):... print(name, fullname)ed Ed Joneswendy Wendy Williamsmary Mary Contraryfred Fred Flinstone由Query返回的元组是命名的元组,提供的KeyedTuple类,并可以像普通的 Python 对象多处理。名称是相同的属性,该属性的名称和类的类名:>>> for row in session.query(User, User.name).all():... print(row.User, row.name)ed wendy mary fred
filter_by() 它使用关键字参数:
>>> for name, in session.query(User.name).\... filter_by(fullname='Ed Jones'):... print(name)ed
filter_by接收的参数形式是关键字参数,而filter接收的参数是更加灵活的SQL表达式结构:
sqlalchemy源码对filter_by的定义def filter_by(self, **kwargs):举例:for user in session.query(User).filter_by(name=’ed’).all(): print userfor user in session.query(User).filter(User.name==”ed”).all(): print user
query.filter(User.name == 'ed')
query.filter(User.name !='ed')
query.filter(User.name.like('%d%')
query.filter(User.name.in_(['a','b','c'])
query.filter(~User.name.in_(['ed','x'])
filter(User.name==None)
filter(User.name!=None)
from sqlalchemy import and_filter(and_(User.name == 'ed',User.fullname=='xxx'))
或者多次调用filter或filter_by
filter(User.name =='ed').filter(User.fullname=='xx')
还可以是:
query.filter(User.name == ‘ed’, User.fullname == ‘Ed Jones’)
from sqlalchemy import or_query.filter(or_(User.name == ‘ed’, User.name == ‘wendy’))
此外,filter函数还可以接收text对象,text是SQL查询语句的字面对象,比如:
for user in session.query(User).filter(text(“id<224”)).order_by(text(“id”)).all(): print user.name
有两种count,第一种是纯粹是执行SQL语句后返回有多少行,对应的函数count(),第二个是func.count(),适用在分组统计,比如按性别分组时,男的有多少,女的多少:
session.query(User).filter(User.name==’ed’).count()session.query(func.count(), User.name).group_by(User.name).all( )
SQLAlchemy中的映射关系有四种,分别是一对多,多对一,一对一,多对多
因为外键(ForeignKey)始终定义在多的一方.如果relationship定义在多的一方,那就是多对一,一对多与多对一的区别在于其关联(relationship)的属性在多的一方还是一的一方,如果relationship定义在一的一方那就是一对多.
这里的例子中,一指的是Parent,一个parent有多个child:class Parent(Base): __tablename__ = 'parent' id = Column(Integer,primary_key = True) children = relationship("Child",backref='parent')class Child(Base): __tablename__ = 'child' id = Column(Integer,primary_key = True) parent_id = Column(Integer,ForeignKey('parent.id'))
这个例子中many是指parent了,意思是一个child可能有多个parent(父亲和母亲),这里的外键(child_id)和relationship(child)都定义在多(parent)的一方:
class Parent(Base): __tablename__ = 'parent' id = Column(Integer, primary_key=True) child_id = Column(Integer, ForeignKey('child.id')) child = relationship("Child", backref="parents")class Child(Base): __tablename__ = 'child' id = Column(Integer, primary_key=True)
为了建立双向关系,可以在relationship()中设置backref,Child对象就有parents属性.设置 cascade= ‘all’,可以级联删除:
class Parent(Base): __tablename__ = 'parent' id = Column(Integer,primary_key = True) children = relationship("Child",cascade='all',backref='parent')def delete_parent(): session = Session() parent = session.query(Parent).get(2) session.delete(parent) session.commit()
不过不设置cascade,删除parent时,其关联的chilren不会删除,只会把chilren关联的parent.id置为空,设置cascade后就可以级联删除children
一对一就是多对一和一对多的一个特例,只需在relationship加上一个参数uselist=False替换多的一端就是一对一:
从一对多转换到一对一:class Parent(Base): __tablename__ = 'parent' id = Column(Integer, primary_key=True) child = relationship("Child", uselist=False, backref="parent")class Child(Base): __tablename__ = 'child' id = Column(Integer, primary_key=True)parent_id = Column(Integer, ForeignKey('parent.id'))
从多对一转换到一对一:
class Parent(Base): __tablename__ = 'parent' id = Column(Integer, primary_key=True) child_id = Column(Integer, ForeignKey('child.id')) child = relationship("Child", backref=backref("parent", uselist=False))class Child(Base): __tablename__ = 'child' id = Column(Integer, primary_key=True)
多对多关系需要一个中间关联表,通过参数secondary来指定,
from sqlalchemy import Table,Textpost_keywords = Table('post_keywords',Base.metadata, Column('post_id',Integer,ForeignKey('posts.id')), Column('keyword_id',Integer,ForeignKey('keywords.id')))class BlogPost(Base): __tablename__ = 'posts' id = Column(Integer,primary_key=True) body = Column(Text) keywords = relationship('Keyword',secondary=post_keywords,backref='posts')class Keyword(Base): __tablename__ = 'keywords' id = Column(Integer,primary_key = True) keyword = Column(String(50),nullable=False,unique=True)
简单地可以使用:
>>> for u, a in session.query(User, Address).\... filter(User.id==Address.user_id).\... filter(Address.email_address=='jack@google.com').\... all():... print(u)... print(a)
如果是使用真正的关联SQL语法来查询可以使用:
>>> session.query(User).join(Address).\... filter(Address.email_address=='jack@google.com').\... all()[]
因为这里的外键就一个,系统知道如何去关联
relationship():函数接收的参数非常多,比如:backref,secondary,primaryjoin,等等。列举一下我用到的参数:
backref:在一对多或多对一之间建立双向关系,比如:class Parent(Base): __tablename__ = 'parent' id = Column(Integer, primary_key=True) children = relationship("Child", backref="parent")class Child(Base): __tablename__ = 'child' id = Column(Integer, primary_key=True) parent_id = Column(Integer, ForeignKey('parent.id'))
Prarent对象获取children,parent.children,反过来Child对象可以获取parent:child.parent.
lazy:默认值是True,说明关联对象只有到真正访问的时候才会去查询数据库,比如有parent对象,只有知道访问parent.children的时候才做关联查询.
remote_side:表中的外键引用的是自身时,如Node类,如果想表示多对一的关系,那么就可以使用remote_side
class Node(Base): __tablename__ = 'node' id = Column(Integer, primary_key=True) parent_id = Column(Integer, ForeignKey('node.id')) data = Column(String(50)) parent = relationship("Node", remote_side=[id])
如果是想建立一种双向的关系,那么还是结合backref:
class Node(Base):__tablename__ = 'node'id = Column(Integer, primary_key=True)parent_id = Column(Integer, ForeignKey('node.id'))data = Column(String(50))children = relationship("Node", backref=backref('parent', remote_side=[id]) )
primaryjoin:用在一对多或者多对一的关系中,默认情况连接条件就是主键与另一端的外键,用primaryjoin参数可以用来指定连接条件 ,比如:下面user的address必须现address是一’tony’开头:
class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True) name = Column(String) addresses = relationship("Address", primaryjoin="and_(User.id==Address.user_id, " "Address.email.startswith('tony'))", backref="user")class Address(Base): __tablename__ = 'address' id = Column(Integer, primary_key=True) email = Column(String) user_id = Column(Integer, ForeignKey('user.id'))secondary:
order_by:
在一对多的关系中,如下代码:class User(Base):....addresses = relationship("Address", order_by="desc(Address.email)", primaryjoin="Address.user_id==User.id")
如果user的address要按照email排序,那么就可以在relationship中添加参数order_by.这里的参数是一字符串形式表示的,不过它等同于python表达式,其实还有另一种基于lambda的方式:
class User(Base):...addresses = relationship(lambda: Address, order_by=lambda: desc(Address.email), primaryjoin=lambda: Address.user_id==User.id)
#!/usr/bin/env python# -*- coding: UTF-8 -*-from sqlalchemy.orm import mapper, sessionmaker__author__ = 'jpzhang' from sqlalchemy import create_engine, Table, Column, Integer, String, MetaDatafrom sqlalchemy.sql.expression import Castfrom sqlalchemy.ext.compiler import compilesfrom sqlalchemy.dialects.mysql import \ BIGINT, BINARY, BIT, BLOB, BOOLEAN, CHAR, DATE, \ DATETIME, DECIMAL, DECIMAL, DOUBLE, ENUM, FLOAT, INTEGER, \ LONGBLOB, LONGTEXT, MEDIUMBLOB, MEDIUMINT, MEDIUMTEXT, NCHAR, \ NUMERIC, NVARCHAR, REAL, SET, SMALLINT, TEXT, TIME, TIMESTAMP, \ TINYBLOB, TINYINT, TINYTEXT, VARBINARY, VARCHAR, YEAR#表的属性描述对象metadata = MetaData()userTable = Table( "wzp_user",metadata, Column('user_id', Integer, primary_key=True), Column('user_name', VARCHAR(50), unique=True, nullable=False), Column('password', VARCHAR(40), nullable=True))#创建数据库连接,MySQLdb连接方式mysql_db = create_engine('mysql://用户名:密码@ip:port/dbname')#创建数据库连接,使用 mysql-connector-python连接方式#mysql_db = create_engine("mysql+mysqlconnector://用户名:密码@ip:port/dbname")#生成表metadata.create_all(mysql_db)#创建一个映射类class User(object): pass#把表映射到类mapper(User, userTable)#创建了一个自定义了的 Session类Session = sessionmaker()#将创建的数据库连接关联到这个sessionSession.configure(bind=mysql_db)session = Session()def main(): u = User() #给映射类添加以下必要的属性,因为上面创建表指定这个字段不能为空,且唯一 u.user_name='tan9le测试' #按照上面创建表的相关代码,这个字段允许为空 u.password='123456' #在session中添加内容 session.add(u) #保存数据 session.flush() #数据库事务的提交,sisson自动过期而不需要关闭 session.commit() #query() 简单的理解就是select() 的支持 ORM 的替代方法,可以接受任意组合的 class/column 表达式 query = session.query(User) #列出所有user print list(query) #根据主键显示 print query.get(1) #类似于SQL的where,打印其中的第一个 print query.filter_by(user_name='tan9le测试').first() u = query.filter_by(user_name='tan9le测试').first() #修改其密码字段 u.password = '654321' #提交事务 session.commit() #打印会出现新密码 print query.get(1).password #根据id字段排序,打印其中的用户名和密码 for instance in session.query(User).order_by(User.user_id): print instance.user_name, instance.password #释放资源 session.close()if __name__ == '__main__': main()