本文共 3886 字，大约阅读时间需要 12 分钟。

SQLAlchemy是一个强大的Python SQL工具箱, 提供了包括ORM在内的各种支持.

首先使用pip安装;

pip install SQLAlchemy

SQL Expression Language 是SQLAlchemy Core的组成部分, 提供了与SQL类似的API而避免了直接书写SQL语句.

连接数据库

使用下列语句连接数据库:

from sqlalchemy import create_engineengine = create_engine("mysql://root:passwd@localhost:3306/webpy?charset=utf8", echo=True)

create_engine的第一个参数的格式为:

数据库类型://用户名:密码@数据库主机地址/数据库名?编码

没有密码则为空，不填; echo=True则在终端进行回显, 默认为False.

engine可以直接执行sql语句:

engine.execute("SELECT * FROM user")

如果需要事务性操作则需要使用connect:

conn = engine.connect()result = conn.execute("SELECT * FROM user")

Create Table

使用metadata和Table类来定义表:

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKeyimport MySQLdbengine = create_engine("mysql://root:password@localhost:3306/test?charset=utf8", encoding="utf-8", echo=True)metadata = MetaData()# defineuser = Table('user', metadata,        Column('user_id', Integer, primary_key=True),        Column('name', String(20)),    )# do create and commitmetadata.create_all(engine)

回显信息:

...INFO sqlalchemy.engine.base.Engine CREATE TABLE user (	user_id INTEGER NOT NULL AUTO_INCREMENT, 	name VARCHAR(20), 	PRIMARY KEY (user_id))....INFO sqlalchemy.engine.base.Engine COMMIT

Insert

使用engine.connect()来执行:

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKeyimport MySQLdbengine = create_engine("mysql://root:password@localhost:3306/test?charset=utf8", encoding="utf-8", echo=True)metadata = MetaData()user = Table('user', metadata,        Column('user_id', Integer, primary_key=True),        Column('name', String(20)),        Column('fullname', String(40)),    )conn = engine.connect()i = user.insert()u = {'name': 'yichya'}result = conn.execute(i, **u)

select

仍然使用conn对象, 先做一个select * from user

conn = engine.connect()s = user.select()result = conn.execute(s)print (result.fetchall() )

回显:

INFO sqlalchemy.engine.base.Engine SELECT user.user_id, user.name, user.fullname  FROM userINFO sqlalchemy.engine.base.Engine ()[(1L, u'yichya', None)]

可以使用for-each遍历结果集:

>>>for raw in result:...   print(raw)INFO sqlalchemy.engine.base.Engine ()(1L, u'yichya', None)(2L, u'finley', None)(3L, u'yy', None)(4L, u'godess', None)

使用user的column对象可以自定义查询的字段

>>>print(user.c)['user.user_id', 'user.name', 'user.fullname']

注意select不是从user对象中获得的:

from sqlalchemy.sql import selectconn = engine.connect()s = select([user.c.user_id, user.c.name])result = conn.execute(s)print (result.fetchall() )

可以在终端看到:

INFO sqlalchemy.engine.base.Engine SELECT user.user_id, user.name FROM userINFO sqlalchemy.engine.base.Engine ()[(1L, u'yichya'), (2L, u'finley'), (3L, u'yy'), (4L, u'godess')]

使用where()筛选记录:

from sqlalchemy.sql import selectconn = engine.connect()s = select([user.c.name]).where(user.c.name == 'yy')result = conn.execute(s)print(result.fetchall() )

可以看到回显:

INFO sqlalchemy.engine.base.Engine SELECT user.name FROM user WHERE user.name = %s INFO sqlalchemy.engine.base.Engine ('yy',)[(u'yy',)]

实际上user.c重载了各种运算符:

>>>print(user.c.name == None)"user".name IS NULL

where支持的运算符包括: > < <= >= == 和 != , 逻辑符and,or, not需要用下述方法:

>>>from sqlalchemy.sql import and_>>>print(and_(user.c.user_id == address.c.user_id, address.c.city == 'New York') )"user".user_id = address.user_id AND address.city = :city_1

顺便展示了多表连接查询, 也可以使用join进行查询:

>>>s = select([user.c.name, address.c.email]).select_from(user.join(address, user.c.user_id==address.c.user_id))>>>print sSELECT "user".name, address.email FROM "user" JOIN address ON "user".user_id = address.user_id

更多的查询请参见,文末的参考链接.

Update

仍然使用connect执行:

conn = engine.connect()u = user.update().where(user.name == 'yichya').values(name='aa')result = conn.execute(u)

Delete

delete也很简单:

>>>d = user.delete().where(user.c.name == 'yichya')>>>conn.execute(d)INFO sqlalchemy.engine.base.Engine DELETE FROM user WHERE user.name = %sINFO sqlalchemy.engine.base.Engine ('yichya',)INFO sqlalchemy.engine.base.Engine COMMIT

参考资料:

转载地址：http://foqzz.baihongyu.com/