前言:最近開始學(xué)習(xí)SQLAlchemy,本教程是其官方文檔以及在讀英文版<Essential SQLAlchemy>的翻譯加一些自己的理解和總結(jié)

1 什么是 SQLAlchemy?

它是給mysql, oracle,sqlite等關(guān)系型數(shù)據(jù)庫的python接口,不需要大幅修改原有的python代碼,它已經(jīng)包含了SQL表達式語言和ORM,看一些例子:

sql=”INSERT INTO user(user_name, password) VALUES (%s, %s)”
cursor = conn.cursor()
cursor.execute(sql, (‘dongwm’, ‘testpass’))

以上是一個常用的mysql的SQL語句,但是冗長也容易出錯,并且可能導(dǎo)致安全問題(因為是字符串的語句,會存在SQL注入),并且代碼不跨平臺,在不同數(shù)據(jù)庫軟件的語句不同(以下是一個 Oracle例子),不具備客移植性:

sql=”INSERT INTO user(user_name, password) VALUES (:1, :2)”
cursor = conn.cursor()
cursor.execute(sql, ‘dongwm’, ‘testpass’)

而在SQLAlchemy里只需要這樣寫:

statement = user_table.insert(user_name=’rick’, password=’parrot’)
statement.execute()  #護略是什么數(shù)據(jù)庫環(huán)境

SQLAlchemy還能讓你寫出很pythonic的語句:

statement = user_table.select(and_(
user_table.c.created >= date(2007,1,1),
user_table.c.created < date(2008,1,1))
result = statement.execute()  #檢索所有在2007年創(chuàng)建的用戶

metadata=MetaData(‘sqlite://’) # 告訴它你設(shè)置的數(shù)據(jù)庫類型是基于內(nèi)存的sqlite
user_table = Table(  #創(chuàng)建一個表
‘tf_user’, metadata,
Column(‘id’, Integer, primary_key=True),  #一些字段,假設(shè)你懂SQL,那么以下的字段很好理解
Column(‘user_name’, Unicode(16), unique=True, nullable=False),
Column(‘email_address’, Unicode(255), unique=True, nullable=False),
Column(‘password’, Unicode(40), nullable=False),
Column(‘first_name’, Unicode(255), default=”),
Column(‘last_name’, Unicode(255), default=”),
Column(‘created’, DateTime, default=datetime.now))

users_table = Table(‘users’, metadata, autoload=True) #假設(shè)table已經(jīng)存在.就不需要指定字段,只是加個autoload=True

class User(object): pass  #雖然SQLAlchemy強大,但是插入更新還是需要手動指定,可以使用ORM,方法就是:設(shè)定一個類,定義一個表,把表映射到類里面
mapper(User, user_table)

下面是一個完整ORM的例子:

from sqlalchemy.orm import mapper, sessionmaker #sessionmaker()函數(shù)是最常使用的創(chuàng)建最頂層可用于整個應(yīng)用Session 的方法,Session管理著所有與數(shù)據(jù)庫之間的會話

from datetime import datetime

from sqlalchemy import Table, MetaData, Column,ForeignKey, Integer, String, Unicode, DateTime #會SQL的人能理解這些函數(shù)吧?

engine = create_engine("sqlite:///tutorial.db",echo=True) #創(chuàng)建到數(shù)據(jù)庫的連接,echo=True表示用logging輸出調(diào)試結(jié)果

metadata = MetaData() #跟蹤表屬性

user_table = Table( #創(chuàng)建一個表所需的信息:字段,表名等

'tf_user', metadata,

Column('id', Integer, primary_key=True),

Column('user_name', Unicode(16), unique=True,nullable=False),

Column('email_address', Unicode(255), unique=True,nullable=False),

Column('password', Unicode(40), nullable=False),

Column('first_name', Unicode(255), default=''),

Column('last_name', Unicode(255), default=''),

Column('created', DateTime, default=datetime.now))

metadata.create_all(engine)  #在數(shù)據(jù)庫中生成表

class User(object): pass #創(chuàng)建一個映射類

mapper(User, user_table) #把表映射到類

Session = sessionmaker() #創(chuàng)建了一個自定義了的 Session類

Session.configure(bind=engine)  #將創(chuàng)建的數(shù)據(jù)庫連接關(guān)聯(lián)到這個session

session = Session()

u = User()

u.user_name='dongwm'

u.email_address='dongwm@dongwm.com'

u.password='testpass' #給映射類添加以下必要的屬性,因為上面創(chuàng)建表指定這幾個字段不能為空

session.add(u) #在session中添加內(nèi)容

session.flush() #保存數(shù)據(jù)

session.commit() #數(shù)據(jù)庫事務(wù)的提交,sisson自動過期而不需要關(guān)閉

query = session.query(User) #query() 簡單的理解就是select() 的支持 ORM 的替代方法,可以接受任意組合的 class/column 表達式

print list(query) #列出所有user

print query.get(1) #根據(jù)主鍵顯示

print query.filter_by(user_name='dongwm').first()  #類似于SQL的where,打印其中的第一個

u = query.filter_by(user_name='dongwm').first()

u.password = 'newpass' #修改其密碼字段

session.commit() #提交事務(wù)

print query.get(1).password #打印會出現(xiàn)新密碼

for instance insession.query(User).order_by(User.id):  #根據(jù)id字段排序,打印其中的用戶名和郵箱地址

    printinstance.user_name, instance.email_address

既然是ORM框架,我們來一個更復(fù)雜的包含關(guān)系的例子,先看sql語句:

CREATE TABLE tf_user (
id INTEGER NOT NULL,
user_name VARCHAR(16) NOT NULL,
email_address VARCHAR(255) NOT NULL,
password VARCHAR(40) NOT NULL,
first_name VARCHAR(255),
last_name VARCHAR(255),
created TIMESTAMP,
PRIMARY KEY (id),
UNIQUE (user_name),
UNIQUE (email_address));
CREATE TABLE tf_group (
id INTEGER NOT NULL,
group_name VARCHAR(16) NOT NULL,
PRIMARY KEY (id),
UNIQUE (group_name));
CREATE TABLE tf_permission (
id INTEGER NOT NULL,
permission_name VARCHAR(16) NOT NULL,
PRIMARY KEY (id),
UNIQUE (permission_name));
CREATE TABLE user_group (
user_id INTEGER,
group_id INTEGER,
PRIMARY KEY(user_id, group_id),
FOREIGN KEY(user_id) REFERENCES tf_user (id), #user_group的user_id關(guān)聯(lián)了tf_user的id字段
FOREIGN KEY(group_id) REFERENCES tf_group (id));  #group_id關(guān)聯(lián)了 tf_group 的id字段

CREATE TABLE group_permission (
group_id INTEGER,
permission_id INTEGER,
PRIMARY KEY(group_id, permission_id),
FOREIGN KEY(group_id) REFERENCES tf_group (id),  #group_permission的id關(guān)聯(lián) tf_group的id字段
FOREIGN KEY(permission_id) REFERENCES tf_permission (id)); #permission_id關(guān)聯(lián)了tf_permission 的id字段

這是一個復(fù)雜的多對多的關(guān)系,比如檢查用戶是否有admin權(quán)限,sql需要這樣:

SELECT COUNT(*) FROM tf_user, tf_group, tf_permission WHERE
tf_user.user_name=’dongwm’ AND tf_user.id=user_group.user_id
AND user_group.group_id = group_permission.group_id
AND group_permission.permission_id = tf_permission.id
AND permission_name=’admin’;  看起來太復(fù)雜并且繁長了

在面向?qū)ο蟮氖澜缋?是這樣的:

class User(object):
groups=[]
class Group(object):
users=[]
permissions=[]
class Permission(object):
groups=[]

print 'Summary for %s' % user.user_name

for g in user.groups:

    print 'Member of group %s' % g.group_name

    for p ing.permissions:

        print'... which has permission %s' % p.permission_name

def user_has_permission(user, permission_name):  #檢查用戶是否有permission_name的權(quán)限的函數(shù)

    for g inuser.groups:

        for p ing.permissions: #可以看出來使用了for循環(huán)

            ifp.permission_name == 'admin':

               return True

    return False

而在SQLAlchemy中,這樣做:

mapper(User, user_table, properties=dict(
groups=relation(Group, secondary=user_group, backref=’users’))) #properties是一個字典值。增加了一個groups 值，它又是一個relation 對象，這個對象實現(xiàn)

#了Group類與user_group的  映射。這樣我通過user_table的groups 屬性就可以反映出RssFeed的值來,

#中間表對象(user_group)傳給secondary參數(shù)，backref為自己的表(users)
mapper(Group, group_table, properties=dict(
permissions=relation(Permission, secondary=group_permission,
backref=’groups’)))
mapper(Permission, permission_table)

q = session.query(Permission)
dongwm_is_admin = q.count_by(permission_name=’admin’,user_name=’dongwm’)

假如計算組里用戶數(shù)(不包含忘記刪除但是重復(fù)的)

for p in permissions:
users = set()
for g in p.groups:
for u in g.users:
users.add(u)
print ‘Permission %s has %d users’ % (p.permission_name, len(users))

在SQLAlchemy可以這樣:

q=select([Permission.c.permission_name,
func.count(user_group.c.user_id)],
and_(Permission.c.id==group_permission.c.permission_id,
Group.c.id==group_permission.c.group_id,
Group.c.id==user_group.c.group_id),
group_by=[Permission.c.permission_name],
distinct=True)
rs=q.execute()
for permission_name, num_users in q.execute():
print ‘Permission %s has %d users’ % (permission_name, num_users) #雖然也長,但是減少了數(shù)據(jù)庫查詢次數(shù),也就是讓簡單事情簡單化,復(fù)雜事情可能簡單解決

看一個綜合的例子:

class User(object):  #這些類設(shè)計數(shù)據(jù)庫的模型

def __init__(self, group_name=None, users=None, permissions=None):
if users is None: users = []
if permissions is None: permissions = []
self.group_name = group_name
self._users = users
self._permissions = permissions

def add_user(self, user):
self._users.append(user)

def del_user(self, user):
self._users.remove(user)

def add_permission(self, permission):
self._permissions.append(permission)

def del_permission(self, permission):
self._permissions.remove(permission)

class Permission(object):

def __init__(self, permission_name=None, groups=None):
self.permission_name = permission_name
self._groups = groups

def join_group(self, group):
self._groups.append(group)

def leave_group(self, group):
self._groups.remove(group)

用sqlalchemy的效果是這樣的:

user_table = Table(
‘tf_user’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘user_name’, Unicode(16), unique=True, nullable=False),
Column(‘password’, Unicode(40), nullable=False))

group_table = Table(
‘tf_group’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘group_name’, Unicode(16), unique=True, nullable=False))

permission_table = Table(
‘tf_permission’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘permission_name’, Unicode(16), unique=True,
nullable=False))

user_group = Table(
‘user_group’,  metadata,
Column(‘user_id’, None, ForeignKey(‘tf_user.id’),
primary_key=True),
Column(‘group_id’, None, ForeignKey(‘tf_group.id’),
primary_key=True))

group_permission = Table(
‘group_permission’,  metadata,
Column(‘group_id’, None, ForeignKey(‘tf_group.id’),
primary_key=True),
Column(‘permission_id’, None, ForeignKey(‘tf_permission.id’),
primary_key=True))

mapper(User, user_table, properties=dict(
_groups=relation(Group, secondary=user_group, backref=’_users’)))
mapper(Group, group_table, properties=dict(
_permissions=relation(Permission, secondary=group_permission,
backref=_’groups’)))
mapper(Permission, permission_table)

這里沒有修改對象,而join_group,leave_group這樣的函數(shù)依然可用,sqlalchemy會跟蹤變化,并且自動刷新數(shù)據(jù)庫

上面介紹了一個完整的例子,連接數(shù)據(jù)庫嗨可以這樣:

engine = create_engine(‘sqlite://’)
connection = engine.connect()  #使用connect
result = connection.execute(“select user_name from tf_user”)
for row in result:
print ‘user name: %s’ % row['user_name']
result.close()

engine = create_engine(‘sqlite://’, strategy=’threadlocal’) #,strategy=’threadlocal’表示重用其它本地線程減少對數(shù)據(jù)庫的訪問

from sqlalchemy.databases.mysql import MSEnum, MSBigInteger  #這個 sqlalchemy.databases是某數(shù)據(jù)庫軟件的’方言’集合,只支持特定平臺
user_table = Table(‘tf_user’, meta,
Column(‘id’, MSBigInteger),
Column(‘honorific’, MSEnum(‘Mr’, ‘Mrs’, ‘Ms’, ‘Miss’, ‘Dr’, ‘Prof’)))

以下是幾個MetaData的應(yīng)用:
unbound_meta = MetaData()  #這個metadata沒有綁定
db1 = create_engine(‘sqlite://’)
unbound_meta.bind = db1  #關(guān)聯(lián)引擎

db2 = MetaData(‘sqlite:///test1.db’)  #直接設(shè)置引擎
bound_meta1 = MetaData(db2)

# Create a bound MetaData with an implicitly created engine
bound_meta2 = MetaData(‘sqlite:///test2.db’)  #隱式綁定引擎
meta = MetaData(‘sqlite://’) #直接綁定引擎可以讓源數(shù)據(jù)直接訪問數(shù)據(jù)庫

user_table = Table(
‘tf_user’, meta,
Column(‘id’, Integer, primary_key=True),
Column(‘user_name’, Unicode(16), unique=True, nullable=False),
Column(‘password’, Unicode(40), nullable=False))

group_table = Table(
‘tf_group’, meta,
Column(‘id’, Integer, primary_key=True),
Column(‘group_name’, Unicode(16), unique=True, nullable=False))

meta.create_all() #創(chuàng)建所有的數(shù)據(jù)庫(以上2個),函數(shù)無參數(shù)

result_set = group_table.select().execute() #選取 group_table的所有表數(shù)據(jù)

以下看一個關(guān)聯(lián)多引擎的例子:

meta = MetaData()  #這里不能直接關(guān)聯(lián)了
engine1 = create_engine(‘sqlite:///test1.db’)  #2個引擎
engine2 = create_engine(‘sqlite:///test2.db’)

# Use the engine parameter to load tables from the first engineuser_table= Table(
‘tf_user’, meta, autoload=True, autoload_with=engine1)  #從第一個引擎加載這些表
group_table = Table(
‘tf_group’, meta, autoload=True, autoload_with=engine1)
permission_table = Table(
‘tf_permission’, meta, autoload=True, autoload_with=engine1)
user_group_table = Table(
‘user_group’, meta, autoload=True, autoload_with=engine1)
group_permission_table = Table(
‘group_permission’, meta, autoload=True, autoload_with=engine1)

meta.create_all(engine2) #在第二個引擎里面創(chuàng)建表

class ImageType(sqlalchemy.types.Binary):  #自定義我們的table的類
def convert_bind_param(self, value, engine):
sfp = StringIO()
value.save(sfp, ‘JPEG’)
return sfp.getvalue()
def convert_result_value(self, value, engine):
sfp = StringIO(value)
image = PIL.Image.open(sfp)
return image  #這里我們定義了一個圖形處理的類型

當(dāng)定義了metadata后,會自定生成一個table.cobject:
q = user_table.select(  #查詢創(chuàng)建在2007年6月1號之前的用戶,并且第一個字母是’r’
user_table.c.user_name.like(‘r%’)  #這里的c就是那個特殊的類,當(dāng)使用sql表達式會用到
& user_table.c.created < datetime(2007,6,1))
或者替代這樣:
q = user_table.select(and_(
user_table.c.user_name.like(‘r%’),
user_table.c.created < datetime(2007,6,1)))
也可以使用rom映射:
q = session.query(User)
q = q.filter(User.c.user_name.like(‘r%’)
& User.c.created > datetime(2007,6,1))

還是一個ORM的例子:

user_table = Table(
‘tf_user’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘user_name’, Unicode(16), unique=True, nullable=False),
Column(‘email_address’, Unicode(255), unique=True, nullable=False),
Column(‘password’, Unicode(40), nullable=False),
Column(‘first_name’, Unicode(255), default=”),
Column(‘last_name’, Unicode(255), default=”),
Column(‘created’, DateTime, default=datetime.now))  #這是一個定義的表類型

group_table = Table(
‘tf_group’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘group_name’, Unicode(16), unique=True, nullable=False))

user_group = Table(
‘user_group’,  metadata,
Column(‘user_id’, None, ForeignKey(‘tf_user.id’), primary_key=True),
Column(‘group_id’, None, ForeignKey(‘tf_group.id’),
… primary_key=True))

import sha
class User(object):  #映射類

def _get_password(self):
return self._password
def _set_password(self, value):
self._password = sha.new(value).hexdigest() #只存儲用戶的哈希密碼
password=property(_get_password, _set_password)

def password_matches(self, password):
return sha.new(password).hexdigest() == self._password

mapper(User, user_table, properties=dict( #映射將創(chuàng)建id, user_name, email_address, password, first_name, last_name,created等字段
_password=user_table.c.password)) #使用哈希后的密碼替換真實密碼,數(shù)據(jù)庫只保存哈希后的,這里在orm上修改

mapper(User, user_table, properties=dict(
_password=user_table.c.password,
groups=relation(Group, secondary=user_group, backref=’users’))) #這里表示可以訪問所有的組，用戶只需訪問一個成員團體屬性,user_group映射類添加group和Group關(guān)聯(lián),

# User類添加users訪問group屬性,看效果:
group1.users.append(user1)  #給group1添加用戶user1,自動更新
user2.groups.append(group2) #把user2添加到group2組,自動更新

對于SQLAlchemy的一些總結(jié):

1 metadata.create_all()

創(chuàng)建多個table可以這樣使用,但是他還有個功能,它添加了”IF NOT EXISTS”,就是在數(shù)據(jù)庫存在的時候,他還是安全的

2 交互模式下的一個全過程:

dongwm@localhost ~ $ python

Python 2.7.3 (default, Jul 11 2012, 10:10:17)

[GCC 4.5.3] on linux2

Type "help", "copyright", "credits" or"license" for more information.

>>> from sqlalchemy import create_engine

>>> from sqlalchemy import Table, MetaData, Column, ForeignKey,Integer, String, Unicode, DateTime

>>> from datetime import datetime

>>> metadata = MetaData('sqlite:///tutorial.db')

>>> user_table = Table(

...     'tf_user', metadata,

...     Column('id', Integer,primary_key=True),

...     Column('user_name',Unicode(16),

...            unique=True,nullable=False),

...     Column('password',Unicode(40), nullable=False),

...     Column('display_name',Unicode(255), default=''),

...     Column('created', DateTime,default=datetime.now))

__main__:7: SAWarning: Unicode column received non-unicode default value.

>>> stmt = user_table.insert() #插入數(shù)據(jù)

>>> stmt.execute(user_name='dongwm1',password='secret',display_name='testdongwm1')

/usr/lib/python2.7/site-packages/SQLAlchemy-0.7.8-py2.7-linux-i686.egg/sqlalchemy/engine/default.py:463:SAWarning: Unicode type received non-unicode bind param value.

 param.append(processors[key](compiled_params[key]))

<sqlalchemy.engine.base.ResultProxy object at 0x8377fcc>

>>> stmt.execute(user_name='dongwm2',password='secret',display_name='testdongwm2') #這個實例可以多次插入,和sql區(qū)別很大

<sqlalchemy.engine.base.ResultProxy object at 0x837e4ec>

>>> stmt = user_table.select() #select查詢

>>> result = stmt.execute()

>>> for row in result:

...     print row

...

(1, u'dongwm1', u'secret', u'testdongwm1', datetime.datetime(2012, 7, 17,11, 57, 48, 515953))

(2, u'dongwm2', u'secret', u'testdongwm2', datetime.datetime(2012, 7, 17,11, 58, 5, 226977))

>>> result = stmt.execute()

>>> row =result.fetchone() #只獲取符合要求的第一項

>>> print  row['user_name']

dongwm1

>>> print row.password

secret

>>> print row.items()

[(u'id', 1), (u'user_name', u'dongwm1'), (u'password', u'secret'),(u'display_name', u'testdongwm1'), (u'created', datetime.datetime(2012, 7, 17,11, 57, 48, 515953))]

>>> stmt = user_table.select(user_table.c.user_name=='dongwm1') #過濾留下user_name=='dongwm1的項

>>> print stmt.execute().fetchall() #獲取所有符合項

[(1, u'dongwm1', u'secret', u'testdongwm1', datetime.datetime(2012, 7, 17,11, 57, 48, 515953))]

>>> stmt = user_table.update(user_table.c.user_name=='dongwm1') #更新數(shù)據(jù)

>>> stmt.execute(password='secret123') #修改密碼

<sqlalchemy.engine.base.ResultProxy object at 0x8377f6c>

>>> stmt = user_table.delete(user_table.c.user_name !='dongwm1')  #刪除user_name不是dongwm1的條目

>>> stmt.execute()

<sqlalchemy.engine.base.ResultProxy object at 0x837f3ac>

>>> user_table.select().execute().fetchall() #查詢發(fā)現(xiàn)就剩一條了

[(1, u'dongwm1', u'secret123', u'testdongwm1', datetime.datetime(2012, 7,17, 11, 57, 48, 515953))]

3 sission上面已經(jīng)說過了,補充一些:

session.delete(u) #把映射類從會話中刪除

4 關(guān)于引擎

引擎就是根據(jù)不同的數(shù)據(jù)庫方言連接數(shù)據(jù)庫的方法

以下是一些例子(方法 driver://username:password@host:port/database):
engine = create_engine(‘sqlite://’)  #連接基于內(nèi)存的sqlite
engine = create_engine(‘sqlite:///data.sqlite’)  #連接基于硬盤文件的sqlite
engine = create_engine(‘postgres://dongwm:foo@localhost:5432/pg_db’)  #連接postgresql
engine = create_engine(‘mysql://localhost/mysql_db’)  #連接mysql
engine = create_engine(‘oracle://dongwm:foo@oracle_tns’) #連接基于TNS協(xié)議的Oracle
engine =create_engine(‘oracle://dongwm:foo@localhost:1521/oracle_sid’) #連接沒有TNS名字的Oracle

也可以帶一些參數(shù):

url=’postgres://dongwm:foo@localhost/pg_db?arg1=foo&arg2=bar’
engine = create_engine(url)

或者:

engine = create_engine(‘postgres://dongwm:foo@localhost/pg_db’,
connect_args=dict(arg1=’foo’, arg2=’bar’))

還可以通過函數(shù)完全控制連接:

import psycopg
def connect_pg():
return psycopg.connect(user=’rick’, host=’localhost’)
engine = create_engine(‘postgres://’, creator=connect_pg)

import logging
handler = logging.FileHandler(‘sqlalchemy.engine.log’)  #可以給它添加一個日志文件處理類
handler.level = logging.DEBUG
logging.getLogger(‘sqlalchemy.engine’).addHandler(handler)

上面說的操作表,也可以直接操作數(shù)據(jù)庫:

conn = engine.connect()
result = conn.execute(‘select user_name, email_address from tf_user’) #結(jié)果是一個sqlalchemy.engine.ResultProxy的實例
for row in result:
print ‘User name: %s Email address: %s’ % (
row['user_name'], row['email_address'])
conn.close()

from sqlalchemy import pool #本來它已經(jīng)自動通過數(shù)據(jù)庫連接管理數(shù)據(jù)池,但是也可以手動管理
import psycopg2
psycopg = pool.manage(psycopg2) #結(jié)果是一個sqlalchemy.pool.DBProxy實例
connection = psycopg.connect(database=’mydb’,
username=’rick’, password=’foo’)

5 關(guān)于元數(shù)據(jù)metadata

它收集了描述table對象等的元數(shù)據(jù)類,當(dāng)使用ORM等時必須使用metadata

如果他被綁定了,那么使用table.create()就會生成表,沒有綁定需要:table.create(bind=some_engine_or_connection),其中table.create

包含一些函數(shù):

autoload:默認是false,當(dāng)數(shù)據(jù)庫已經(jīng)存在這個table會自動加載覆蓋

autoload_with:默認是false,是否自動加載引擎的字段結(jié)構(gòu)

reflect:默認是false,是否體現(xiàn)源表結(jié)構(gòu)

brand_table = Table(‘brand’, metadata,
Column(‘name’, Unicode(255)), # 覆蓋類型
autoload=True)

6 關(guān)于表結(jié)構(gòu):

設(shè)置表主鍵可以這樣:

Column(‘brand_id’, Integer, ForeignKey(‘brand.id’),primary_key=True), #通過primary_key=True
Column(‘sku’, Unicode(80), primary_key=True))
也可以這樣:
product_table = Table(
‘product’, metadata,
Column(‘brand_id’, Integer, ForeignKey(‘brand.id’)),
Column(‘sku’, Unicode(80)),
PrimaryKeyConstraint(‘brand_id’, ‘sku’, name=’prikey’))  #通過PrimaryKeyConstraint

style_table = Table(
‘style’, metadata,
Column(‘brand_id’, Integer, primary_key=True),
Column(‘sku’, Unicode(80), primary_key=True),
Column(‘code’, Unicode(80), primary_key=True),
ForeignKeyConstraint(  #使用復(fù)合鍵,關(guān)聯(lián)外部表的字段
['brand_id', 'sku'],
['product.brand_id', 'product.sku']))

product_table = Table(
‘product’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘brand_id’, Integer, ForeignKey(‘brand.id’)), #他的brand_id關(guān)聯(lián)brand的讓id
Column(‘sku’, Unicode(80)),
UniqueConstraint(‘brand_id’, ‘sku’)) #約束唯一標(biāo)識數(shù)據(jù)庫表中的每條記錄

payment_table = Table(
‘payment’, metadata,
Column(‘a(chǎn)mount’, Numeric(10,2), CheckConstraint(‘a(chǎn)mount > 0′))) #驗證amount大于0
user_table = Table(
‘tf_user’, MetaData(),
Column(‘id’, Integer, primary_key=True),
Column(‘user_name’, Unicode(16), unique=True, nullable=False),
Column(‘password’, Unicode(40), nullable=False),
Column(‘first_name’, Unicode(255), default=”),
Column(‘last_name’, Unicode(255), default=”),
Column(‘created_apptime’, DateTime, default=datetime.now), #default表示當(dāng)不舍定具體值時設(shè)定一個默認值
Column(‘created_dbtime’, DateTime, PassiveDefault(‘sysdate’)), # PassiveDefault是數(shù)據(jù)庫級別的默認值,
Column(‘modified’, DateTime, onupdate=datetime.now)) #單設(shè)置onupdate這個屬性，這是不應(yīng)用到數(shù)據(jù)庫的設(shè)計中的.只是存在于映射類中.

#它是活躍更新的,因為每次執(zhí)行的時間都不同

user_table = Table(
‘tf_user’, MetaData(),
Column(‘id’, Integer, primary_key=True),
Column(‘user_name’, Unicode(16), unique=True, nullable=False, index=True), #一旦數(shù)據(jù)庫增長到一定規(guī)模時，可能需要考慮增加表的索引，以加快某些操作
Column(‘password’, Unicode(40), nullable=False),
Column(‘first_name’, Unicode(255), default=”),
Column(‘last_name’, Unicode(255), default=”, index=True))

其中指定索引也可以這樣:

i = Index(‘idx_name’,user_table.c.first_name,user_table.c.last_name,unique=True)
i.create(bind=e)

brand_table = Table(
‘brand’, metadata,
Column(‘id’, Integer, Sequence(‘brand_id_seq’), primary_key=True),  #需要通過序列化方式來創(chuàng)建新主鍵標(biāo)識符的數(shù)據(jù)庫，

#SQLAlchemy 并不會自動為其生成?？梢灾付⊿equence生成
Column(‘name’, Unicode(255), unique=True, nullable=False))

7 元數(shù)據(jù)操作

meta1 = MetaData(‘postgres://postgres:password@localhost/test’,
… reflect=True)
meta2 = MetaData(‘sqlite://’)
for table in meta1.table_iterator():
table.tometadata(meta2) #通過這個方法讓meta1的元數(shù)據(jù)被meta2使用
meta2.create_all()

2 假如想放棄綁定使用drop_all( )或者drop(e)

1 自定義表結(jié)構(gòu)類型:

from sqlalchemy import types

class MyCustomEnum(types.TypeDecorator):  #自定義的類型繼承至types.TypeDecorator

impl=types.Integer  #實現(xiàn)指定的類型int

def __init__(self, enum_values, *l, **kw):
types.TypeDecorator.__init__(self, *l, **kw)
self._enum_values = enum_values

def convert_bind_param(self, value, engine):  #必須含有這個方法,轉(zhuǎn)換python語言為SQL
result = self.impl.convert_bind_param(value, engine)
if result not in self._enum_values:
raise TypeError, (
“Value %s must be one of %s” % (result, self._enum_values))
return result

def convert_result_value(self, value, engine):  #必須含有這個方法,通過db的api把SQL轉(zhuǎn)換成python語言
‘Do nothing here’
return self.impl.convert_result_value(value, engine)

看一個例子:

from sqlalchemy import types

from sqlalchemy.databases import sqlite

class MyCustomEnum(types.TypeDecorator):

    impl = types.Integer

    def __init__(self, enum_values,*l, **kw):

       types.TypeDecorator.__init__(self, *l, **kw)

        self._enum_values =enum_values

    def bind_processor(self,dialect): #如果提供這個方法會替代convert_bind_param( )和convert_result_value( )

        impl_processor =self.impl.bind_processor(dialect)

        if impl_processor:

            def processor(value):

                result =impl_processor(value)

                assert value inself._enum_values, \

                    "Value %smust be one of %s" % (result,

                   self._enum_values)

                return result

        else:

            def processor(value):

                assert value inself._enum_values, \

                    "Value %smust be one of %s" % (value,

                     self._enum_values)

                return value

        return processor

mce=MyCustomEnum([1,2,3])

processor = mce.bind_processor(sqlite.dialect())

print processor(1) #返回1

print processor(5) #返回錯誤,因為不是1,2,3中的數(shù)據(jù)

你甚至可以直接定義自定的TypeDecorator

class NewType(types.TypeEngine): #TypeDecorator繼承自types.TypeEngine

def __init__(self, *args):
self._args = args

def get_col_spec(self):  #create_table( )會用到這個方法
return ‘NEWTYPE(%s)’ % ‘,’.join(self._args)

def convert_bind_param(self, value, engine):  #這個必須設(shè)置
return value

def convert_result_value(self, value, engine):  #這個也必須設(shè)置
return value

2 SQL語句在交互模式下的例子:

dongwm@localhost ~ $ python
Python 2.7.3 (default, Jul 11 2012, 10:10:17)
[GCC 4.5.3] on linux2
Type “help”, “copyright”, “credits” or “l(fā)icense” for more information.
>>> from sqlalchemy import Table, MetaData, Column, ForeignKey,Integer, String, Unicode, DateTime
>>> metadata=MetaData()
>>> simple_table = Table(  #一個簡單的表結(jié)構(gòu)
…     ‘simple’, metadata,
…     Column(‘id’, Integer, primary_key=True),
…     Column(‘col1′, Unicode(20)))
>>>
>>> stmt = simple_table.insert()  #插入數(shù)據(jù)操作的實例
>>> print stmt #打印這個實例
INSERT INTO simple (id, col1) VALUES (:id, :col1) #里面包含需要替換的變量
>>> compiled_stmt = stmt.compile()  #編譯語句
>>> print compiled_stmt.params #轉(zhuǎn)成了字典得方式
{‘id’: None, ‘col1′: None}
>>> from sqlalchemy import create_engine
>>> engine = create_engine(‘sqlite://’)
>>> simple_table.create(bind=engine)  #創(chuàng)建table
>>> engine.execute(stmt, col1=”Foo”) #給語句添加值
/usr/lib/python2.7/site-packages/SQLAlchemy-0.7.8-py2.7-linux-i686.egg/sqlalchemy/engine/default.py:463:SAWarning: Unicode type received non-unicode bind param value.
param.append(processors[key](compiled_params[key]))
<sqlalchemy.engine.base.ResultProxy object at 0x8376c8c>
>>> metadata.bind = engine  #和上面效果一樣,給語句添加值
>>> stmt.execute(col1=”Bar”)
<sqlalchemy.engine.base.ResultProxy object at 0x8376f4c>
>>> stmt = simple_table.insert(values=dict(col1=”Initial value”)) #這次插入已經(jīng)設(shè)置了值
>>> print stmt
INSERT INTO simple (col1) VALUES (?)
>>> compiled_stmt = stmt.compile()
>>> print compiled_stmt.params
{‘col1′: ‘Initial value’}
>>> stmt = simple_table.insert()
>>> stmt.execute(col1=”First value”)
<sqlalchemy.engine.base.ResultProxy object at 0x838832c>
>>>
>>> stmt.execute(col1=”Second value”)
<sqlalchemy.engine.base.ResultProxy object at 0x838844c>
>>> stmt.execute(col1=”Third value”) #這樣一行一行插入真是費勁
<sqlalchemy.engine.base.ResultProxy object at 0x838856c>
>>> stmt.execute([dict(col1="Fourth Value"), #可以一次插入多行
...     dict(col1="Fifth Value"),
...     dict(col1="Sixth Value")])
<sqlalchemy.engine.base.ResultProxy object at 0x83886ac>
>>> from sqlalchemy import text
>>> stmt = simple_table.update(
…     whereclause=text(“col1=’First value’”),
…     values=dict(col1=’1st Value’))  #執(zhí)行col1是First value的條目設(shè)置值為1st Value
>>> stmt.execute()
<sqlalchemy.engine.base.ResultProxy object at 0x838878c>
>>> stmt = simple_table.update(text(“col1=’Second value’”)) #尋找col1是Second value的條目
>>> stmt.execute(col1=’2nd Value’) #執(zhí)行更新時,設(shè)置其值,想過和上面的一樣
<sqlalchemy.engine.base.ResultProxy object at 0x8376f4c>
>>> stmt = simple_table.update(text(“col1=’Third value’”))
>>> print stmt
UPDATE simple SET id=?, col1=? WHERE col1=’Third value’
>>> engine.echo = True #設(shè)置打印調(diào)試日志
>>> stmt.execute(col1=’3rd value’)
2012-07-17 15:16:59,231 INFO sqlalchemy.engine.base.Engine UPDATE simple SETcol1=? WHERE col1=’Third value’
2012-07-17 15:16:59,245 INFO sqlalchemy.engine.base.Engine (’3rd value’,)
2012-07-17 15:16:59,245 INFO sqlalchemy.engine.base.Engine COMMIT
<sqlalchemy.engine.base.ResultProxy object at 0x83767ec>

>>> stmt = simple_table.delete(  #刪除
…     text(“col1=’Second value’”))
>>> stmt.execute()
2012-07-17 15:21:03,806 INFO sqlalchemy.engine.base.Engine DELETE FROM simpleWHERE col1=’Second value’
2012-07-17 15:21:03,806 INFO sqlalchemy.engine.base.Engine ()
2012-07-17 15:21:03,806 INFO sqlalchemy.engine.base.Engine COMMIT
<sqlalchemy.engine.base.ResultProxy object at 0x8376a0c>
>>> from sqlalchemy import select
>>> stmt = select([simple_table.c.col1])  #查詢col1這個字段
>>> for row in stmt.execute():
…     print row
(u’Foo’,)
(u’Bar’,)
(u’1st Value’,)
(u’2nd Value’,)
(u’3rd value’,)
(u’Fourth Value’,)
(u’Fifth Value’,)
(u’Sixth Value’,)

>>> stmt = simple_table.select() #和上面的區(qū)別是這是條目全部顯示
>>> for row in stmt.execute():  #這2句也可以這樣表示stmt = select( simple_table])
…     print row
…
(1, u’Foo’)
(2, u’Bar’)
(3, u’1st Value’)
(4, u’2nd Value’)
(5, u’3rd value’)
(6, u’Fourth Value’)
(7, u’Fifth Value’)
(8, u’Sixth Value’)
>>> x = simple_table.c.col1==”Foo”
>>> print type(x)
<class ‘sqlalchemy.sql.expression._BinaryExpression’>
>>> print x
simple.col1 = :col1_1
>>> expr = simple_table.c.col1 + “-col1″  #它還支持運算符
>>> print expr
simple.col1 || :col1_1
>>> from sqlalchemy.databases import mysql
>>> print expr.compile(dialect=mysql.MySQLDialect())
concat(simple.col1, %s) #在不同的數(shù)據(jù)庫軟件,效果不同

>>> from sqlalchemy import func
>>> print func.now()
now()
>>> print func.current_timestamp
<sqlalchemy.sql.expression._FunctionGenerator object at 0x83888cc>
>>> print func._(text(‘a(chǎn)=b’))
(a=b)

注:sqlalchemy支持in,op,startwith,endwith,between,like等運算

>>> from sqlalchemy import bindparam  #自定義綁定的詞
>>> stmt = select([simple_table.c.col1],
…    whereclause=simple_table.c.col1==bindparam(‘test’))  #用test替換原來的col1
>>> print stmt
SELECT simple.col1
FROM simple
WHERE simple.col1 = ? #這里依然是col1
>>> print stmt.execute(test=’Foo’).fetchall()
[(u'Foo',)]

>>> stmt =simple_table.select(order_by=[simple_table.c.col1])  #更具col1,升序排序
>>> print stmt
SELECT simple.id, simple.col1
FROM simple ORDER BY simple.col1
>>> print stmt.execute().fetchall()
[(3, u'1st Value'), (4, u'2nd Value'), (5, u'3rd value'), (2, u'Bar'), (7,u'Fifth Value'), (1, u'Foo'), (6, u'Fourth Value'), (8, u'Sixth Value')]
>>> from sqlalchemy import desc
>>> stmt = simple_table.select(order_by=[desc(simple_table.c.col1)]) #根據(jù)col1,降序排序
>>> print stmt
SELECT simple.id, simple.col1
FROM simple ORDER BY simple.col1 DESC
>>> print stmt.execute().fetchall()
[(8, u'Sixth Value'), (6, u'Fourth Value'), (1, u'Foo'), (7, u'Fifth Value'),(2, u'Bar'), (5, u'3rd value'), (4, u'2nd Value'), (3, u'1st Value')]
注:distinct=True去重復(fù),效果類似于SELECT DISTINCT

>>> stmt = simple_table.select(offset=1, limit=1) #offset設(shè)置偏移,這里就是略過第一個,返回第二個.limit設(shè)置返回多少個條目
>>> print stmt
SELECT simple.id, simple.col1
FROM simple
LIMIT ? OFFSET ?
>>> print stmt.execute().fetchall()
[(2, u'Bar')]
看下面的例子:

“Persons” 表：

“Orders” 表：

現(xiàn)在，我們希望列出所有的人，以及他們的定購號碼:

SELECT Persons.LastName, Persons.FirstName,Orders.OrderNo

FROM Persons

LEFT JOIN Orders #將orders表join進來

ON Persons.Id_P=Orders.Id_P #關(guān)系聯(lián)系

ORDER BY Persons.LastName #排序

書中的例子是這樣的:

SELECT store.name

FROM store

JOIN product_price ON store.id=product_price.store_id

JOIN product ON product_price.sku=product.sku

WHERE product.msrp != product_price.price;

轉(zhuǎn)換成sqlalchemy語句:

>>>from_obj =store_table.join(product_price_table).join(product_table)

>>> query = store_table.select()

>>> query = query.select_from(from_obj)

>>> query = query.where(product_table.c.msrp !=product_price_table.c.price)

>>> print query

SELECT store.id, store.name

FROM store JOIN product_price ON store.id =product_price.store_id JOIN product ON product.sku = product_price.sku

WHERE product.msrp != product_price.price

>>> print query.column(‘product.sku’)
SELECT store.id, store.name, product.sku
FROM store JOIN product_price ON store.id = product_price.store_id  JOINproduct ON product.sku = product_price.sku
WHERE product.msrp != product_price.price
>>> query2 = select([store_table,product_table.c.sku],from_obj=[from_obj],whereclause=(product_table.c.msrp!=product_price_table.c.price))
>>> print query2
SELECT store.id, store.name, product.sku
FROM store JOIN product_price ON store.id = product_price.store_id JOIN productON product.sku = product_price.sku
WHERE product.msrp != product_price.price
>>> query = product_table.select(and_(product_table.c.msrp > 10.00,product_table.c.msrp < 20.00)) #范圍查詢
>>> print query
SELECT product.sku, product.msrp
FROM product
WHERE product.msrp > ? AND product.msrp < ?
>>> for r in query.execute():
…print r
(u’123′, Decimal(“12.34″))

>>> from sqlalchemy import intersect

>>> query0 = product_table.select(product_table.c.msrp >10.00)
>>> query1 = product_table.select(product_table.c.msrp < 20.00)
>>> query = intersect(query0, query1) #使用 intersect添加多query
>>> print query
SELECT product.sku, product.msrp

employee_table = Table(
‘employee’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘manager’, None, ForeignKey(‘employee.id’)),
Column(‘name’, String(255)))

給設(shè)定alias:

比如想實現(xiàn)以下SQL

SELECT employee.name
FROM employee, employee AS manager
WHERE employee.manager_id = manager.id
AND manager.name = ‘Fred’

>>> manager = employee_table.alias(‘mgr’)
>>> stmt = select([employee_table.c.name],
…
and_(employee_table.c.manager_id==manager.c.id,
…
manager.c.name==’Fred’))
>>> print stmt
SELECT employee.name
FROM employee, employee AS mgr
WHERE employee.manager_id = mgr.id AND mgr.name = ?

>>> manager = employee_table.alias()  #自動alias
>>> stmt = select([employee_table.c.name],
…and_(employee_table.c.manager_id==manager.c.id,
…manager.c.name==’Fred’))
>>> print stmt
SELECT employee.name
FROM employee, employee AS employee_1
WHERE employee.manager_id = employee_1.id AND employee_1.name = ?

from sqlalchemy import types
class MyCustomEnum(types.TypeDecorator):
impl=types.Integer
def __init__(self, enum_values, *l, **kw):
types.TypeDecorator.__init__(self, *l, **kw)
self._enum_values = enum_values
def convert_bind_param(self, value, engine):
result = self.impl.convert_bind_param(value, engine)
if result not in self._enum_values:
raise TypeError, (
“Value %s must be one of %s” % (result, self._enum_values))
Application-Specific Custom Types | 63return result
def convert_result_value(self, value, engine):
‘Do nothing here’
return self.impl.convert_result_value(value, engine)

1  ORM模型的簡單性簡化了數(shù)據(jù)庫查詢過程。使用ORM查詢工具，用戶可以訪問期望數(shù)據(jù)，而不必理解數(shù)據(jù)庫的底層結(jié)構(gòu)

以下是SQL語句:
region_table = Table(
‘region’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘name’, Unicode(255)))
相應(yīng)的類:
class Region(object):

def __init__(self, name):
self.name = name

def __repr__(self):
return ‘<Region %s>’ % self.name

看一下在交互模式下:

>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__']
>>> mapper(Region,region_table)  #ORM映射
<Mapper at 0x84bdb2c; Region>
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__', '_sa_class_manager', 'id','name'] #增加了很多屬性
>>> Region.id
<sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x84c238c>
>>> Region.name
<sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x84c254c>

>>> r0 = Region(name=”Northeast”)
>>> r1 = Region(name=”Southwest”)
>>> r0
<Region Northeast>  #類能顯示這樣的數(shù)據(jù)是因為類定義了__repr__方法
>>> r1
<Region Southwest>
>>> from sqlalchemy.orm import clear_mappers
>>> clear_mappers() #取消映射
>>> Region.name #不再有這個屬性
Traceback (most recent call last):
File “<stdin>”, line 1, in <module>
AttributeError: type object ‘Region’ has no attribute ‘name’
>>> dir(Region)  #回到了原來的只有類屬性
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__']

>>> r0 = Region(name=”Northeast”)  #從這里開始理解ORM做了什么
>>> r1 = Region(name=”Southwest”) #實現(xiàn)了2個類的實例

>>> metadata.create_all(engine) #創(chuàng)建table

>>> Session = sessionmaker()  #通過sessionmaker產(chǎn)生一個會話
>>> Session.configure(bind=engine) #綁定到數(shù)據(jù)庫連接
>>> session = Session()  #產(chǎn)生會話實例,讓對象可以被載入或保存到數(shù)據(jù)庫,而只需要訪問類卻不用直接訪問數(shù)據(jù)庫
>>> session.bind.echo = True #顯示打印信息

>>> session.add(r1) #把r0,r12個實例加到會話中
>>> session.add(r0)
>>> print r0.id  #因為還沒有保存,數(shù)據(jù)為空
None
>>> session.flush() #提交數(shù)據(jù)到數(shù)據(jù)庫
2012-07-18 10:24:07,116 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-07-18 10:24:07,116 INFO sqlalchemy.engine.base.Engine INSERT INTO region(name) VALUES (?)
2012-07-18 10:24:07,116 INFO sqlalchemy.engine.base.Engine (‘Southwest’,)
2012-07-18 10:24:07,117 INFO sqlalchemy.engine.base.Engine INSERT INTO region(name) VALUES (?)
2012-07-18 10:24:07,117 INFO sqlalchemy.engine.base.Engine (‘Northeast’,)
>>> print r0.id #id因為子增長,出現(xiàn)了
2
>>> r0.name = ‘Northwest’
>>> session.flush() #修改提交
2012-07-18 10:24:50,644 INFO sqlalchemy.engine.base.Engine UPDATE region SETname=? WHERE region.id = ?
2012-07-18 10:24:50,644 INFO sqlalchemy.engine.base.Engine (‘Northwest’, 2)
>>> print r0.name #數(shù)據(jù)庫中的數(shù)據(jù)被update成了新值
Northwest
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__', '__getattribute__','__hash__', '__init__', '__module__', '__new__', '__reduce__', '__reduce_ex__','__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__','__weakref__']
>>> mapper(Region, region_table, include_properties=['id']) #使用 include_properties只映射某些字段,同樣還有exclude_properties
<Mapper at 0x84c26cc; Region>
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__', '_sa_class_manager', 'id'] #只多了一個”id”

>>> clear_mappers()
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__']
>>> mapper(Region, region_table, column_prefix=’_')  #映射后自定義修改新屬性的前綴
<Mapper at 0x84f73ac; Region>
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__', '_id', '_name','_sa_class_manager'] #id和name等前面都有了”_”

>>> clear_mappers()
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__']
>>> mapper(Region, region_table, properties=dict(
…     region_name=region_table.c.name,  #想把name的屬性定義為region_name,因為c.name就是用Table創(chuàng)建的表結(jié)構(gòu)的特定實例的name屬性
…     region_id=region_table.c.id))
<Mapper at 0x8509d2c; Region>
>>> dir(Region)
['__class__', '__delattr__', '__dict__', '__doc__', '__format__','__getattribute__', '__hash__', '__init__', '__module__', '__new__','__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__','__str__', '__subclasshook__', '__weakref__', '_sa_class_manager', 'region_id','region_name']  #id改名為region_id

>>> class Region(object):  #重新定義類
…     def __init__(self, name):
…         self.name = name
…     def __repr__(self):
…         return ‘<Region %s>’ %self.name
…     def _get_name(self): #這個_get和_set是為了讓內(nèi)置的property調(diào)用
…         return self._name
…     def _set_name(self, value):
…         assertvalue.endswith(‘Region’), \
…            ‘Region names must end in “Region”‘
…         self._name = value
…     name=property(_get_name, _set_name) #通過property的定義，當(dāng)獲取成員x的值時，就會調(diào)用_get_name函數(shù)(第一個參數(shù))，當(dāng)給成員x賦值時，就會調(diào)用_set_name函數(shù)(第二個參數(shù))，當(dāng)刪除x時，就會調(diào)用delx函數(shù)(這里沒有設(shè)置)
…
>>> from sqlalchemy.orm import synonym
>>> mapper(Region, region_table, column_prefix=’_', properties=dict(
…     name=synonym(‘_name’))) #首先檢驗_name的屬性是否滿足
<Mapper at 0x84f7acc; Region>
>>> s0 = Region(‘Southeast’)  #沒有正確結(jié)尾
Traceback (most recent call last):
File “<stdin>”, line 1, in <module>
File “<string>”, line 4, in __init__
File“/usr/lib/python2.7/site-packages/SQLAlchemy-0.7.8-py2.7-linux-i686.egg/sqlalchemy/orm/state.py”,line 98, in initialize_instance
return manager.original_init(*mixed[1:], **kwargs)
File “<stdin>”, line 3, in __init__
File “<string>”, line 1, in __set__
File “<stdin>”, line 10, in _set_name
AssertionError: Region names must end in “Region”
>>> s0 = Region(‘Southeast Region’) #正常

>>> segment_table = Table(
…     ‘segment’, metadata,
…     Column(‘id’, Integer, primary_key=True),
…     Column(‘lat0′, Float),
…     Column(‘long0′, Float),
…     Column(‘lat1′, Float),
…     Column(‘long1′, Float))

>>> metadata.create_all(engine) #創(chuàng)建表
>>> class RouteSegment(object): #一個含有begin和end的類
…     def __init__(self, begin, end):
…         self.begin = begin
…         self.end = end
…     def __repr__(self):
…         return ‘<Route %s to%s>’ % (self.begin, self.end)
…
>>> class MapPoint(object):
…     def __init__(self, lat, long):
…         self.coords = lat, long
…     def __composite_values__(self): #返回比較后的列表或者元祖
…         return self.coords
…     def __eq__(self, other):
…         return self.coords ==other.coords
…     def __ne__(self, other):
…         return self.coords !=other.coords
…     def __repr__(self):
…         return ‘(%s lat, %s long)’ %self.coords
…
…

>>> from sqlalchemy.orm import composite
>>> mapper(RouteSegment, segment_table, properties=dict(
…     begin=composite(MapPoint,  #創(chuàng)建多個屬性
…         segment_table.c.lat0,
…         segment_table.c.long0),
…     end=composite(MapPoint,
…         segment_table.c.lat1,segment_table.c.long1)))
<Mapper at 0x86203cc; RouteSegment>
>>> work=MapPoint(33.775562,-84.29478)
>>> library=MapPoint(34.004313,-84.452062)
>>> park=MapPoint(33.776868,-84.389785)
>>> routes = [
...     RouteSegment(work, library),
...     RouteSegment(work, park),
...     RouteSegment(library, work),
...     RouteSegment(library, park),
...     RouteSegment(park, library),
...     RouteSegment(park, work)]

>>> for rs in routes:
…     session.add(rs)
…
>>> session.flush()
>>> q = session.query(RouteSegment)
>>> print RouteSegment.begin==work
segment.lat0 = :lat0_1 AND segment.long0 = :long0_1
>>> q = q.filter(RouteSegment.begin==work)
>>> for rs in q:
…     print rs
…
2012-07-18 11:12:29,360 INFO sqlalchemy.engine.base.Engine SELECT segment.id ASsegment_id, segment.lat0 AS segment_lat0, segment.long0 AS segment_long0,segment.lat1 AS segment_lat1, segment.long1 AS segment_long1
FROM segment
WHERE segment.lat0 = ? AND segment.long0 = ?
2012-07-18 11:12:29,360 INFO sqlalchemy.engine.base.Engine (33.775562,-84.29478)
<Route (33.775562 lat, -84.29478 long) to (34.004313 lat, -84.452062long)>
<Route (33.775562 lat, -84.29478 long) to (33.776868 lat, -84.389785long)>

>>> from sqlalchemy.orm import PropComparator
>>> class MapPointComparator(PropComparator): #自定義運算符繼承PropComparator類
…     def __lt__(self, other):  #自定義小于運算結(jié)果
…         return and_(*[a<b for a, bin
...            zip(self.prop.columns,
...                other.__composite_values__())])
…
>>> mapper(RouteSegment, segment_table, properties=dict(
…     begin=composite(MapPoint,
…                    segment_table.c.lat0, segment_table.c.long0,
…                    comparator=MapPointComparator),  #定義使用自定義的運算類
…     end=composite(MapPoint,
…                  segment_table.c.lat1, segment_table.c.long1,
…                  comparator=MapPointComparator)))
<Mapper at 0x85b2bac; RouteSegment>
>>> product_table = Table(
… ‘product’, metadata,
… Column(‘sku’, String(20), primary_key=True),
… Column(‘msrp’, Numeric),
… Column(‘image’, BLOB))
>>> from sqlalchemy.orm import deferred
>>> mapper(Product, product_table, properties=dict(
…     image=deferred(product_table.c.image))) #deferred意思是延遲,就是在實現(xiàn) mapper 時，可以指定某些字段是 Deferred 裝入的，這樣象通常一樣取出數(shù)據(jù)時，這些字段并不真正的從數(shù)據(jù)庫中取出，只有在你真正需要時才取出，這樣可以減少資源的占用和提高效率,只有在讀取 image時才會取出相應(yīng)的數(shù)據(jù)
<Mapper at 0x862a40c; Product>

>>> metadata.remove(product_table)  #因為已經(jīng)常見了表,先刪除
>>> product_table = Table(
…     ‘product’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric),
…     Column(‘image1′, Binary),
…     Column(‘image2′, Binary),
…     Column(‘image3′, Binary))

>>> clear_mappers() #已經(jīng)映射,先取消
>>> mapper(Product, product_table, properties=dict(
…     image1=deferred(product_table.c.image1,group=’images’),
…     image2=deferred(product_table.c.image2,group=’images’),
…     image3=deferred(product_table.c.image3,group=’images’))) #Deferred字段可以通過在 properties 中指定 group參數(shù)來表示編組情況。這樣當(dāng)一個組的某個

#字段被取出時， 同組的其它字段均被取出
<Mapper at 0x85b8c4c; Product>

>>> q = product_table.join(  被映射的是join了product_summary_table到product_table的結(jié)果
… product_summary_table,
… product_table.c.sku==product_summary_table.c.sku).alias(‘full_product’)
>>> class FullProduct(object): pass
…
>>> mapper(FullProduct, q)
<Mapper at 0x86709cc; FullProduct>

mapper函數(shù)的一些參數(shù):

always_refresh =False:返回查詢舊會修改內(nèi)存中的值,但是populate_existing優(yōu)先級高

allow_column_override =False:允許關(guān)系屬性將具有相同的名稱定義為一個映射列,否則名稱沖突,產(chǎn)生異常

2 ORM的關(guān)系

1 1:N relations (1對多)

>>> mapper(Store, store_table)
<Mapper at 0x84fba4c; Store>
>>> from sqlalchemy.orm import relation
>>> mapper(Region, region_table, properties=dict(
…     stores=relation(Store))) #讓2個表關(guān)聯(lián),給Region添加一個屬性stores,通過它聯(lián)系Store來修改Store
<Mapper at 0x84f76ac; Region>

>>> r0 = Region(‘test’)

>>> session.add(r0) #先生成一條數(shù)據(jù)
>>> session.commit()
2012-07-18 13:56:26,858 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-07-18 13:56:26,859 INFO sqlalchemy.engine.base.Engine INSERT INTO region(name) VALUES (?)
2012-07-18 13:56:26,859 INFO sqlalchemy.engine.base.Engine (‘test’,)
2012-07-18 13:56:26,859 INFO sqlalchemy.engine.base.Engine COMMIT
>>> rgn = session.query(Region).get(1)  #獲取這條數(shù)據(jù)
2012-07-18 13:56:37,250 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-07-18 13:56:37,251 INFO sqlalchemy.engine.base.Engine SELECT region.id ASregion_id, region.name AS region_name
FROM region
WHERE region.id = ?
2012-07-18 13:56:37,251 INFO sqlalchemy.engine.base.Engine (1,)
>>> s0 = Store(name=’3rd and Juniper’) #創(chuàng)建一個實例
>>> rgn.stores.append(s0) #通過Region的依賴建立新的Store(其中的一個字段region_id值來著region的id字段)
2012-07-18 13:56:51,611 INFO sqlalchemy.engine.base.Engine SELECT store.id ASstore_id, store.region_id AS store_region_id, store.name AS store_name
FROM store
WHERE ? = store.region_id
2012-07-18 13:56:51,611 INFO sqlalchemy.engine.base.Engine (1,)
>>> session.flush() #保存數(shù)據(jù)庫
2012-07-18 13:57:02,131 INFO sqlalchemy.engine.base.Engine INSERT INTO store(region_id, name) VALUES (?, ?)
2012-07-18 13:57:02,131 INFO sqlalchemy.engine.base.Engine (1, ’3rd andJuniper’)
注:假如2個表之間有多個外部依賴關(guān)系,需要使用primaryjoin指定:

mapper(Region, region_table, properties=dict(
stores=relation(Store,
primaryjoin=(store_table.c.region_id  #判斷關(guān)系來著region_id和region的id
==region_table.c.id))))

2 M:N relations(多對多)

上面有SQL語句:我復(fù)制過來:

category_table = Table(
‘category’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘level_id’, None, ForeignKey(‘level.id’)),
Column(‘parent_id’, None, ForeignKey(‘category.id’)),
Column(‘name’, String(20)))
product_table = Table(
‘product’, metadata,
Column(‘sku’, String(20), primary_key=True),
Column(‘msrp’, Numeric))
product_category_table = Table(
‘product_category’, metadata,
Column(‘product_id’, None, ForeignKey(‘product.sku’), primary_key=True),
Column(‘category_id’, None, ForeignKey(‘category.id’), primary_key=True))

可以看出來product_category_table和category_table 是多對多的關(guān)系.

>>> mapper(Category, category_table, properties=dict(
…     products=relation(Product,
…     secondary=product_category_table)))
<Mapper at 0x859c8cc; Category>
>>> mapper(Product, product_table, properties=dict(
…     categories=relation(Category,
…     secondary=product_category_table)))
<Mapper at 0x859c5cc; Product>

>>> r0=Product(’123′,’234′)

>>> session.add(r0)
>>> session.flush()
2012-07-18 14:18:06,599 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-07-18 14:18:06,618 INFO sqlalchemy.engine.base.Engine INSERT INTO product(sku, msrp) VALUES (?, ?)
2012-07-18 14:18:06,618 INFO sqlalchemy.engine.base.Engine (’123′, 234.0)
>>> session.query(Product).get(’123′).categories

>>> clear_mappers()
>>> mapper(Category, category_table, properties=dict(
…     products=relation(Product,secondary=product_category_table,
… primaryjoin=(product_category_table.c.category_id  #primaryjoin是要被映射的表和連接表的條件
…                                   == category_table.c.id),
… secondaryjoin=(product_category_table.c.product_id  #secondaryjoin是連接表和想加入的表的條件
…                                     == product_table.c.sku))))
<Mapper at 0x84ff7cc; Category>
>>> mapper(Product, product_table, properties=dict(
…     categories=relation(Category,secondary=product_category_table,
… primaryjoin=(product_category_table.c.product_id
…                                     == product_table.c.sku),
… secondaryjoin=(product_category_table.c.category_id
…                                       == category_table.c.id))))
<Mapper at 0x859cb8c; Product>
1:1 relations(一對一):特殊的(1:N)
還是上面的SQL:

product_table = Table(
‘product’, metadata,
Column(‘sku’, String(20), primary_key=True),
Column(‘msrp’, Numeric))
product_summary_table = Table(
‘product_summary’, metadata,
Column(‘sku’, None, ForeignKey(‘product.sku’), primary_key=True), #只有一個外聯(lián)到product
Column(‘name’, Unicode(255)),
Column(‘description’, Unicode))

>>> mapper(Product, product_table, properties=dict(
…     summary=relation(ProductSummary)))
KeyboardInterrupt
>>> mapper(ProductSummary, product_summary_table)
<Mapper at 0x84fbe6c; ProductSummary>
>>> mapper(Product, product_table, properties=dict(
…     summary=relation(ProductSummary)))
<Mapper at 0x85bee6c; Product>
>>> prod = session.query(Product).get(’123′)
[]  #product_summary_table因為product_table兒存在,浪費了

>>> mapper(ProductSummary, product_summary_table)
<Mapper at 0x84f7dec; ProductSummary>
>>> mapper(Product, product_table, properties=dict(
…    summary=relation(ProductSummary,uselist=False)))  #使用uselist=False就不會這樣了
<Mapper at 0x860584c; Product>
>>> prod = session.query(Product).get(’123′)
>>> print prod.summary
None
>>> mapper(ProductSummary, product_summary_table)
<Mapper at 0x859ca0c; ProductSummary>
>>> mapper(Product, product_table, properties=dict(
…     summary=relation(ProductSummary, uselist=False,
…     backref=’product’))) #自定義自己表的函數(shù)
<Mapper at 0x860e90c; Product>
>>> prod = session.query(Product).get(’123′)
>>> prod.summary = ProductSummary(name=”Fruit”, description=”Some
… Fruit”)
>>> print prod.summary
<ProductSummary Fruit>
>>> print prod.summary.product #他的屬性就是prod,可就是表本身
<Product 123>
>>> print prod.summary.product is prod
True

>>> mapper(Level, level_table, properties=dict(
…     categories=relation(Category, backref=’level’)))
<Mapper at 0x860590c; Level>
>>> mapper(Category, category_table, properties=dict(
…     products=relation(Product,
…        secondary=product_category_table)))
<Mapper at 0x860ec8c; Category>
>>> mapper(Product, product_table, properties=dict(
…     categories=relation(Category,
…        secondary=product_category_table)))
<Mapper at 0x860e7ec; Product>
>>> lvl = Level(name=’Department’)
>>> cat = Category(name=’Produce’, level=lvl)
>>> session.add(lvl)
>>> session.flush()
2012-07-18 14:44:02,005 INFO sqlalchemy.engine.base.Engine INSERT INTO level(parent_id, name) VALUES (?, ?)
2012-07-18 14:44:02,005 INFO sqlalchemy.engine.base.Engine (None, ‘Department’)
2012-07-18 14:44:02,020 INFO sqlalchemy.engine.base.Engine INSERT INTO category(level_id, parent_id, name) VALUES (?, ?, ?)
2012-07-18 14:44:02,020 INFO sqlalchemy.engine.base.Engine (1, None, ‘Produce’)
>>> prod = session.query(Product).get(’123′)
>>> print prod.categories
[]
>>> print cat.products
2012-07-18 14:44:25,517 INFO sqlalchemy.engine.base.Engine SELECT product.skuAS product_sku, product.msrp AS product_msrp
FROM product, product_category
WHERE ? = product_category.category_id AND product.sku =product_category.product_id
2012-07-18 14:44:25,517 INFO sqlalchemy.engine.base.Engine (1,)
[]
>>> prod.categories.append(cat)
>>> print prod.categories
[<Category Department.Produce>]
>>> print cat.products  #backref自動更新,在多對多的情況,可以使用relation函數(shù)兩次,但是2個屬性沒有保持同步
[]  #解決方法:

>>> mapper(Level, level_table, properties=dict(
…categories=relation(Category, backref=’level’)))
>>> mapper(Category, category_table, properties=dict(
…products=relation(Product, secondary=product_category_table,
… backref=’categories’)))  #在Product也設(shè)置backref,就會保持同步
>>> mapper(Product, product_table)
>>> lvl = Level(name=’Department’)
>>> cat = Category(name=’Produce’, level=lvl)
>>> session.save(lvl)
>>> prod = session.query(Product).get(’123′)
>>> print prod.categories
[]
>>> print cat.products
[]
>>> prod.categories.append(cat)
>>> print prod.categories
[<Category Department.Produce>]
>>>print cat.products
[<Product 123>]

>>> from sqlalchemy.orm import backref
>>> clear_mappers()
>>> mapper(ProductSummary, product_summary_table, properties=dict(
… product=relation(Product,
… backref=backref(‘summary’, uselist=False))))  #還可以使用backref函數(shù)做一樣的事情
<Mapper at 0x860aaec; ProductSummary>
>>> mapper(Product, product_table)
<Mapper at 0x85bee6c; Product>

4 Self-Referential 自我參照映射
level_table = Table(
‘level’, metadata,
Column(‘id’, Integer, primary_key=True),
Column(‘parent_id’, None, ForeignKey(‘level.id’)), #這個外聯(lián)其實還是這個類的id,也就是映射了自己的對象
Column(‘name’, String(20)))
>>> mapper(Level, level_table, properties=dict(
… children=relation(Level))) #不同層次之間的父子關(guān)系,我這里指定得到”子”的屬性
<Mapper at 0x860a66c; Level>
>>> mapper(Level, level_table, properties=dict(
…     children=relation(Level,
…     backref=backref(‘parent’,
…     remote_side=[level_table.c.id]))))  #remote_side指定’子’的id,local side”就是字段parent_id
<Mapper at 0x860e42c; Level>
>>> l0 = Level(‘Gender’)
>>> l1 = Level(‘Department’, parent=l0)
>>> session.add(l0)
>>> session.flush()
2012-07-18 15:07:55,810 INFO sqlalchemy.engine.base.Engine INSERT INTO level(parent_id, name) VALUES (?, ?)
2012-07-18 15:07:55,810 INFO sqlalchemy.engine.base.Engine (None, ‘Gender’) #插入l0,他沒有父級
2012-07-18 15:07:55,810 INFO sqlalchemy.engine.base.Engine INSERT INTO level(parent_id, name) VALUES (?, ?)
2012-07-18 15:07:55,810 INFO sqlalchemy.engine.base.Engine (2, ‘Department’)

注 我們還能反過來用:

mapper(Level, level_table, properties=dict(
parent=relation(Level, remote_side=[level_table.c.parent_id],
backref=’children’)))

我們創(chuàng)建一個多引擎的例子:

from sqlalchemy import create_engine

from sqlalchemy.orm import mapper, sessionmaker

from sqlalchemy import Numeric,Table, MetaData, Column, ForeignKey,Integer, String

engine1 = create_engine('sqlite://')

engine2 = create_engine('sqlite://')

metadata = MetaData()

product_table = Table(

'product', metadata,

Column('sku', String(20), primary_key=True),

Column('msrp', Numeric))

product_summary_table = Table(

'product_summary', metadata,

Column('sku', String(20), ForeignKey('product.sku'), primary_key=True),

Column('name', Unicode(255)),

Column('description', Unicode))

product_table.create(bind=engine1)

product_summary_table.create(bind=engine2)

stmt = product_table.insert()

engine1.execute(

stmt,

[dict(sku="123", msrp=12.34),

dict(sku="456", msrp=22.12),

dict(sku="789", msrp=41.44)])

stmt = product_summary_table.insert()

engine2.execute(

stmt,

[dict(sku="123", name="Shoes", description="SomeShoes"),

dict(sku="456", name="Pants", description="SomePants"),

dict(sku="789", name="Shirts", description="Some Shirts")])

這樣就創(chuàng)建了表并且插入了一些數(shù)據(jù)

dongwm@localhost ~ $ python
Python 2.7.3 (default, Jul 11 2012, 10:10:17)
[GCC 4.5.3] on linux2
Type “help”, “copyright”, “credits” or “l(fā)icense” for more information.
>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import mapper, sessionmaker
>>> from sqlalchemy import Numeric,Table, MetaData, Column,ForeignKey, Integer, String,Unicode
>>> engine1 = create_engine(‘sqlite://’)
>>> engine2 = create_engine(‘sqlite://’)  #創(chuàng)建多個引擎
>>> metadata = MetaData()
>>> product_table = Table(
… ‘product’, metadata,
… Column(‘sku’, String(20), primary_key=True),
… Column(‘msrp’, Numeric))
>>> product_summary_table = Table(
… ‘product_summary’, metadata,
… Column(‘sku’, String(20), ForeignKey(‘product.sku’), primary_key=True),
… Column(‘name’, Unicode(255)),
… Column(‘description’, Unicode))
>>> product_table.create(bind=engine1)
>>> product_summary_table.create(bind=engine2)
>>> stmt = product_table.insert()
>>> engine1.execute(
… stmt,
… [dict(sku="123", msrp=12.34),
... dict(sku="456", msrp=22.12),
... dict(sku="789", msrp=41.44)])
<sqlalchemy.engine.base.ResultProxy object at 0x84ef9ec>
>>> stmt = product_summary_table.insert()
>>> engine2.execute(  #用引擎2 插入數(shù)據(jù),那么product_summary的數(shù)據(jù)就在這個引擎
… stmt,
… [dict(sku="123", name="Shoes", description="SomeShoes"),
... dict(sku="456", name="Pants", description="SomePants"),
... dict(sku="789", name="Shirts", description="SomeShirts")])
/usr/lib/python2.7/site-packages/SQLAlchemy-0.7.8-py2.7-linux-i686.egg/sqlalchemy/engine/default.py:463:SAWarning: Unicode type received non-unicode bind param value.
param.append(processors[key](compiled_params[key]))
<sqlalchemy.engine.base.ResultProxy object at 0x84e896c>
>>> class Product(object):
…     def __init__(self, sku, msrp, summary=None):
…         self.sku = sku
…         self.msrp = msrp
…         self.summary = summary
…     def __repr__(self):
…         return ‘<Product %s>’ %self.sku
…
>>> class ProductSummary(object):
…     def __init__(self, name, description):
…         self.name = name
…         self.description =description
…     def __repr__(self):
…         return ‘<ProductSummary%s>’ % self.name
…
>>> from sqlalchemy.orm import clear_mappers,backref,relation
>>> clear_mappers()
>>> mapper(ProductSummary, product_summary_table, properties=dict(
…     product=relation(Product,
…                     backref=backref(‘summary’, uselist=False))))
<Mapper at 0x84efa4c; ProductSummary>
>>> mapper(Product, product_table)
<Mapper at 0x84efd0c; Product>
>>> Session = sessionmaker(binds={Product:engine1,  #這里綁定了2個引擎,不同orm的引擎不同
…     ProductSummary:engine2})
>>> session = Session()
>>> engine1.echo = engine2.echo = True
>>> session.query(Product).all() #查詢product的數(shù)據(jù)
2012-07-18 19:00:59,514 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-07-18 19:00:59,514 INFO sqlalchemy.engine.base.Engine SELECT product.skuAS product_sku, product.msrp AS product_msrp
FROM product
2012-07-18 19:00:59,514 INFO sqlalchemy.engine.base.Engine ()
/usr/lib/python2.7/site-packages/SQLAlchemy-0.7.8-py2.7-linux-i686.egg/sqlalchemy/types.py:215:SAWarning: Dialect sqlite+pysqlite does *not* support Decimal objects natively,and SQLAlchemy must convert from floating point – rounding errors and otherissues may occur. Please consider storing Decimal numbers as strings orintegers on this platform for lossless storage.
d[coltype] = rp = d['impl'].result_processor(dialect, coltype)
[<Product 123>, <Product 456>, <Product 789>]
>>> session.query(ProductSummary).all() #查詢ProductSummary
2012-07-18 19:01:07,510 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2012-07-18 19:01:07,510 INFO sqlalchemy.engine.base.Engine SELECTproduct_summary.sku AS product_summary_sku, product_summary.name ASproduct_summary_name, product_summary.description AS product_summary_description
FROM product_summary
2012-07-18 19:01:07,510 INFO sqlalchemy.engine.base.Engine ()
[<ProductSummary Shoes>, <ProductSummary Pants>, <ProductSummaryShirts>]

>>> from sqlalchemy.orm.shard import ShardedSession #使用ShardedSession對會話水平分區(qū),根據(jù)需求把數(shù)據(jù)分開
>>> product_table = Table(
…     ‘product’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric))
>>> metadata.create_all(bind=engine1)
>>> metadata.create_all(bind=engine2)
>>> class Product(object):
…     def __init__(self, sku, msrp):
…         self.sku = sku
…         self.msrp = msrp
…     def __repr__(self):
…         return ‘<Product %s>’ %self.sku
…
>>> clear_mappers()
>>> product_mapper = mapper(Product, product_table)
>>> def shard_chooser(mapper, instance, clause=None):  #返回包含映射和實例的行的分區(qū)ID
…     if mapper is not product_mapper: #非設(shè)定的orm映射叫做odd
…         return ‘odd’
…     if (instance.sku  #數(shù)據(jù)為偶數(shù)也叫做even
…         and instance.sku[0].isdigit()
…         and int(instance.sku[0]) % 2== 0):
…         return ‘even’
…     else:
…         return ‘odd’ #否則叫做odd
…

>>> def id_chooser(query, ident):  根據(jù)查詢和映射類的主鍵返回對象想通過查詢駐留的shard ID列表
…     if query.mapper is not product_mapper:
…         return ['odd']
…     if (ident \
…         and ident[0].isdigit()
…         and int(ident[0]) % 2 == 0):
…         return ['even']
…     return ['odd']
…
>>> def query_chooser(query): #返回可選的shard ID列表
…     return ['even', 'odd']
…
>>> Session = sessionmaker(class_=ShardedSession)
>>> session = Session(
…     shard_chooser=shard_chooser,
…     id_chooser=id_chooser,
…     query_chooser=query_chooser,
…     shards=dict(even=engine1,
…                odd=engine2))
>>> products = [ Product('%d%d%d' % (i,i,i), 0.0)
...     for i in range(10) ]
>>> for p in products:
…     session.add(p)
…
>>> session.flush()
>>> for row in engine1.execute(product_table.select()):
…     print row
…
2012-07-18 19:11:19,811 INFO sqlalchemy.engine.base.Engine SELECT product.sku,product.msrp
FROM product
2012-07-18 19:11:19,811 INFO sqlalchemy.engine.base.Engine ()
(u’000′, Decimal(’0E-10′)) #偶數(shù)數(shù)據(jù)寫在engine1
(u’222′, Decimal(’0E-10′))
(u’444′, Decimal(’0E-10′))
(u’666′, Decimal(’0E-10′))
(u’888′, Decimal(’0E-10′))
>>> for row in engine2.execute(product_table.select()):
…     print row
…
2012-07-18 19:11:40,098 INFO sqlalchemy.engine.base.Engine SELECT product.sku,product.msrp
FROM product
2012-07-18 19:11:40,099 INFO sqlalchemy.engine.base.Engine ()
(u’111′, Decimal(’0E-10′)) #奇數(shù)數(shù)據(jù)寫在engine1
(u’333′, Decimal(’0E-10′))
(u’555′, Decimal(’0E-10′))
(u’777′, Decimal(’0E-10′))
(u’999′, Decimal(’0E-10′))
>>> session.query(Product).all()
2012-07-18 19:12:36,130 INFO sqlalchemy.engine.base.Engine SELECT product.skuAS product_sku, product.msrp AS product_msrp
FROM product
2012-07-18 19:12:36,130 INFO sqlalchemy.engine.base.Engine ()
2012-07-18 19:12:36,131 INFO sqlalchemy.engine.base.Engine SELECT product.skuAS product_sku, product.msrp AS product_msrp
FROM product
2012-07-18 19:12:36,131 INFO sqlalchemy.engine.base.Engine ()
[<Product 123>, <Product 456>, <Product 789>, <Product000>, <Product 222>, <Product 444>, <Product 666>,<Product 888>, <Product 111>, <Product 333>, <Product555>, <Product 777>, <Product 999>]

from sqlalchemy import create_engine
from sqlalchemy.orm import mapper, sessionmaker
from datetime import datetime
from sqlalchemy import Numeric,Table, MetaData, Column, ForeignKey, Integer,String, Unicode, DateTime
from sqlalchemy import types
from sqlalchemy.databases import sqlite
engine1 = create_engine(‘sqlite://’)
engine2 = create_engine(‘sqlite://’)
metadata = MetaData()
product_table = Table(
‘product’, metadata,
Column(‘sku’, String(20), primary_key=True),
Column(‘msrp’, Numeric))
product_summary_table = Table(
‘product_summary’, metadata,
Column(‘sku’, String(20), ForeignKey(‘product.sku’), primary_key=True),
Column(‘name’, Unicode(255)),
Column(‘description’, Unicode))
product_table.create(bind=engine1)
product_summary_table.create(bind=engine2)
stmt = product_table.insert()
engine1.execute(
stmt,
[dict(sku="123", msrp=12.34),
dict(sku="456", msrp=22.12),
dict(sku="789", msrp=41.44)])
stmt = product_summary_table.insert()
engine2.execute(
stmt,
[dict(sku="123", name="Shoes", description="SomeShoes"),
dict(sku="456", name="Pants", description="SomePants"),
dict(sku="789", name="Shirts", description="SomeShirts")])

 本文主要說刪除

metadata.drop_all(engine) #刪除某引擎的全部表

metadata.remove(test_table)  #刪除某一個table

clear_mappers() #取消所有的映射

在relation中有一個參數(shù)cascade,它是基于session的操作,包括把對象放入session,從session刪除對象等,如果 指定cascade=”all”表示做的任何session操作給映射類都能很好的工作,默認包含save-update, merge
mapper(ParentClass, parent, properties=dict(
children=relation(ChildClass, backref=’parent’,
cascade=’all,delete-orphan’) )) #delete-orphan表示如果曾經(jīng)是子類(childclass)實例但是卻沒有和父類連接的情況下,假如要刪除這個子類,而不想掛空父類引用了的實例,
額看個例子就懂了:
photo = Table(
… ‘photo’, metadata,
… Column(‘id’, Integer, primary_key=True))
tag = Table(
… ‘tag’, metadata,
… Column(‘id’, Integer, primary_key=True),
… Column(‘photo_id’, None, ForeignKey(‘photo.id’)),
… Column(‘tag’, String(80)))
class Photo(object):
… pass
…
class Tag(object):
… def __init__(self, tag):
… self.tag = tag
…
mapper(Photo, photo, properties=dict(
… tags=relation(Tag, backref=’photo’, cascade=”all”)))
<Mapper at 0x851504c; Photo>
>>> mapper(Tag, tag)
<Mapper at 0x8515dac; Tag>
>>> p1 = Photo()
>>> p2 = Photo()
>>> p1.tags = [Tag(tag='foo'),Tag(tag='bar'),Tag(tag='baz')]
>>> p2.tags = [Tag(tag='foo'),Tag(tag='bar'),Tag(tag='baz')]
>>> session.add(p1)
>>> session.add(p2)
>>> session.flush()
>>> for t in session.query(Tag):
… print t.id,t.photo_id, t.tag
…
1 1 foo #出現(xiàn)以下關(guān)聯(lián)數(shù)據(jù)
2 1 bar
3 1 baz
4 2 foo
5 2 bar
6 2 baz
>>> session.delete(session.query(Photo).get(1)) #刪除一個tag的數(shù)據(jù)
>>> session.flush()
>>> for t in session.query(Tag):
… print t.id, t.photo_id, t.tag
…
4 2 foo #他會刪除關(guān)聯(lián)所有t.photo_id為1的數(shù)據(jù),在這里relation(ChildClass, backref=’parent’,cascade=’all,delete-orphan’)指定delete-orphan沒什么,關(guān)鍵看下面
5 2 bar
6 2 baz
>>> p3 = session.query(Photo).get(2)
>>> del p3.tags[0] #如果我只是刪除關(guān)聯(lián)點…
>>> session.flush()
>>> for t in session.query(Tag):
… print t.id, t.photo_id, t.tag
…
4 None foo #關(guān)聯(lián)點photo_id成了none,但是條目存在 –他不會影響其它關(guān)聯(lián)表
5 2 bar
6 2 baz

>>> p3 = session.query(Photo).get(2) #假如沒有設(shè)置delete-orphan
>>> del p3.tags[0]
>>> session.flush()
>>> for t in session.query(Tag):
… print t.id, t.photo_id, t.tag
5 2 bar #自動刪除了關(guān)聯(lián)的其它表的項
6 2 baz
注:可用的cascade參數(shù)包含:

• save-update -我的理解是調(diào)用session.add()會自動將項目添加到相應(yīng)級聯(lián)關(guān)系上,也適用于已經(jīng)從關(guān)系中刪除的項目嗨沒有來得及刷新的情況
• merge – 它是session.merge的實現(xiàn),復(fù)制狀態(tài)到具有相同標(biāo)識符的持久化實例的實例,如果沒有持久的實例和當(dāng)前session相關(guān)聯(lián)，返回的持久化實例。如果給定的實例未保存，他會保存一個副本，并返回這個副本作為一個新的持久化實例
• expunge – 從session中刪除實例
• delete – 標(biāo)記一個實例被刪除,執(zhí)行flush()會執(zhí)行刪除操作
• delete-orphan-如果子類從母類刪除,標(biāo)記之,但是不影響母類
• refresh-expire – 定期刷新在給定的實例的屬性,查詢并刷新數(shù)據(jù)庫
• all – 以上全部屬性的集合:“save-update,merge, refresh-expire, expunge, delete

·        本文主要是ORM的sission查詢和更新

·        session負責(zé)執(zhí)行內(nèi)存中的對象和數(shù)據(jù)庫表之間的同步工作,創(chuàng)建session可以這樣:

·        Session =sessionmaker(bind=engine) #sqlalchemy.orm.session.Session類有很多參數(shù),使用sessionmaker是為了簡化這個過程

·        或者:
Session = sessionmaker()
Session.configure(bind=engine)

·        注:sessionmaker的參數(shù):
autoflush=True  #為True時，session將在執(zhí)行session的任何查詢前自動調(diào)用flush()。這將確保返回的結(jié)果

·        transactional=False #為True時，session將自動使用事務(wù)commit
twophase=False #當(dāng)處理多個數(shù)據(jù)庫實例,當(dāng)使用flush()但是沒有提交事務(wù)commit時,給每個數(shù)據(jù)庫一個標(biāo)識,使整個事務(wù)回滾

·        創(chuàng)建session,添加數(shù)據(jù)的例子(以前也出現(xiàn)過很多次了)
dongwm@localhost ~ $ python
Python 2.7.3 (default, Jul 11 2012, 10:10:17)
[GCC 4.5.3] on linux2
Type “help”, “copyright”, “credits” or “l(fā)icense” for more information.
>>> from sqlalchemy import *
>>> from sqlalchemy.orm import *
>>> engine = create_engine(‘sqlite://’)
>>> metadata = MetaData(engine)
>>> account_table = Table(
… ‘a(chǎn)ccount’, metadata,
… Column(‘id’, Integer, primary_key=True),
… Column(‘balance’, Numeric))
>>> class Account(object): pass
…
>>> mapper(Account, account_table)
<Mapper at 0x84e6f2c; Account>
>>> account_table.create()
>>> a = Account()
>>> a.balance = 100.00
>>> Session = sessionmaker(bind=engine)
>>> session = Session()
>>> session.add(a)
>>> session.flush()
>>> session.delete(a) #自動刪除 account_table相應(yīng)條目,但是在1:N和M:N關(guān)系中不會自動刪除它的級聯(lián)關(guān)系
>>> session.flush()

·        注:session的對象狀態(tài):

·        Transient:短暫的,主要指內(nèi)存中的對象

·        Pending:掛起的,這樣的對象準備插入數(shù)據(jù)庫,等執(zhí)行了flush就會插入

·        Persistent:持久的

·        Detached:對象在數(shù)據(jù)庫里面有記錄,但是不屬于session

·        >>>make_transient(a)  #因為標(biāo)識了已刪除,恢復(fù)a的狀態(tài)
>>> session.add(a) #重新添加
>>> session.flush()
>>> query = session.query(Account)
>>> print query
SELECT account.id AS account_id, account.balance AS account_balance
FROM account
>>> for obj in query:
…     print obj
…
<__main__.Account object at 0x84eef0c>

·        >>> query.all() #查詢所有
[<__main__.Account object at 0x84eef0c>]
>>> query = query.filter(Account.balance > 10.00)  #filter過濾
>>> for obj in query:
…     print obj.balance
…

·        100.00

·        >>> for i insession.query(Account).filter_by(balance=100.00 ):  #通過條件過濾
…     print i
>>> query = session.query(Account)
>>> query = query.from_statement(‘select *from account wherebalance=:bac’) #通過帶通配符的SQL語句其中:bac標(biāo)識這個參數(shù)是bac
>>> query = query.params(bac=’100.00′) #根據(jù)bac指定值尋找
>>> print query.all()
[<__main__.Account object at 0x84eef0c>]

·        本地session

·        >>> Session =scoped_session(sessionmaker(  #設(shè)置一個本地的共享session
…     bind=engine, autoflush=True))
>>> session = Session()
>>> session2 = Session()
>>> session is session2  #他們是同一個
True

·        >>> a = Account()
>>> a.balance = 100.00
>>> Session.add(a) #注意 這是的’S'是大寫
>>> Session.flush()
>>> b = Account()
>>> a.balance = 200.00
>>> session.add(a)  #其實他們是一個共享的session 名字都可以
>>> session.flush()
>>> print session.query(Account).all() #查詢到了2個
[<__main__.Account object at 0x851be0c>, <__main__.Account object at0x84f7d6c>]

·        注:這樣的映射mapper也可以這樣是用:

·        mapper(Product, product_table,properties=dict(
categories=relation(Category, secondary=product_category_table,
backref=’products’)))

·        Session.mapper(Product,product_table, properties=dict(
categories=relation(Category, secondary=product_category_table,
backref=’products’))) #它的優(yōu)點是可以初始化參數(shù)

·        本文主要是面向?qū)ο蟮睦^承映射到關(guān)系數(shù)據(jù)庫表的方法

·        >>> classProduct(object):
…     def __init__(self, sku, msrp):
…         self.sku = sku
…         self.msrp = msrp
…     def __repr__(self):
…         return ‘<%s %s>’ % (
…            self.__class__.__name__, self.sku)
…
>>> class Clothing(Product):
…     def __init__(self, sku, msrp, clothing_info):
…         Product.__init__(self, sku,msrp) #繼承了Product
…         self.clothing_info =clothing_info
…
>>> class Accessory(Product):
…     def __init__(self, sku, msrp, accessory_info):
…         Product.__init__(self, sku,msrp) #繼承了Product
…         self.accessory_info =accessory_info
也就是這樣的意思:

·        

·        這個單表繼承中(如下圖,黑色的表示沒有被映射):

·        

·        從創(chuàng)建表結(jié)構(gòu)是這樣:

·        >>> product_table =Table(
…     ‘product’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric),
…     Column(‘clothing_info’, String),
…     Column(‘a(chǎn)ccessory_info’, String),
…     Column(‘product_type’, String(1), nullable=False)) #一個新的字段
>>> mapper(
…     Product, product_table,
…     polymorphic_on=product_table.c.product_type, #映射繼承層次結(jié)構(gòu)使用polymorphic_on表示繼承在product_type字段,值是polymorphic_identity指定的標(biāo)識
…     polymorphic_identity=’P') #標(biāo)識繼承 Product ,父類
<Mapper at 0x85833ec; Product>
>>> mapper(Clothing, inherits=Product,
…       polymorphic_identity=’C')   #標(biāo)識繼承Clothingproduct
<Mapper at 0x858362c; Clothing>
>>>
>>> mapper(Accessory, inherits=Product, #繼承至Product
…        polymorphic_identity=’A') #標(biāo)識繼承Accessory
<Mapper at 0x8587d8c; Accessory>
>>> products = [  #創(chuàng)建一些產(chǎn)品
...     Product('123', 11.22),
...     Product('456', 33.44),
...     Clothing('789', 123.45, "Nice Pants"),
...     Clothing('111', 125.45, "Nicer Pants"),
...     Accessory('222', 24.99, "Wallet"),
...     Accessory('333', 14.99, "Belt") ]
>>> Session = sessionmaker()
>>> session = Session()
>>> for p in products:
…     session.add(p)
…
>>> session.flush()
>>> print session.query(Product).all() #全部都有
[<Product 123>, <Product 456>, <Clothing 789>, <Clothing111>, <Accessory 222>, <Accessory 333>]
>>> print session.query(Clothing).all()  #只顯示2個
[<Clothing 789>, <Clothing 111>]
>>> print session.query(Accessory).all() #只顯示2個,是不是上面的映射效果和創(chuàng)建3個類而分別orm好的多呢?
[<Accessory 222>, <Accessory 333>]

·        >>> for row inproduct_table.select().execute(): #從父類庫查詢,所有數(shù)據(jù)都有,只是product_type不同
…     print row
…
(u’123′, Decimal(’11.2200000000′), None, None, u’P')
(u’456′, Decimal(’33.4400000000′), None, None, u’P')
(u’789′, Decimal(’123.4500000000′), u’Nice Pants’, None, u’C')
(u’111′, Decimal(’125.4500000000′), u’Nicer Pants’, None, u’C')
(u’222′, Decimal(’24.9900000000′), None, u’Wallet’, u’A')
(u’333′, Decimal(’14.9900000000′), None, u’Belt’, u’A')

·        具體的映射見下圖:

·        

·        查詢一個沒有的不存在的映射:

·        >>> printsession.query(Accessory)[0].clothing_info
None

·        具體表的繼承

·        每個表包含的數(shù)據(jù)量,需要實現(xiàn)它的類;沒有浪費的空間

·        >>>metadata.remove(product_table)
>>> product_table = Table(
…     ‘product’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric))
>>> clothing_table = Table(
…     ‘clothing’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric),
…     Column(‘clothing_info’, String))
>>>
>>> accessory_table = Table(
…     ‘a(chǎn)ccessory’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric),
…     Column(‘a(chǎn)ccessory_info’, String))
>>>
攝像我們想要獲取Product’sku’是222的數(shù)據(jù)(沒有其他額外的工作),我們不得不層次型的查詢每個類,請看這個例子:

·        >>> punion =polymorphic_union(
…     dict(P=product_table,
…     C=clothing_table,
…     A=accessory_table),
…     ‘type_’)
>>>
>>> print punion
SELECT accessory.sku, accessory.msrp, accessory.accessory_info, CAST(NULL ASVARCHAR) AS clothing_info, ‘A’ AS type_
FROM accessory UNION ALL SELECT product.sku, product.msrp, CAST(NULL ASVARCHAR) AS accessory_info, CAST(NULL AS VARCHAR) AS clothing_info, ‘P’ AStype_
FROM product UNION ALL SELECT clothing.sku, clothing.msrp, CAST(NULL ASVARCHAR) AS accessory_info, clothing.clothing_info, ‘C’ AS type_
FROM clothing
現(xiàn)在我們就有了一個很好的標(biāo)記了(C,A,P)

·        >>> mapper(
…     Product, product_table, with_polymorphic=(‘*’,punion),  #使用with_polymorphic=(‘*’, punion)的方式映射父類,指定不同表選擇,實現(xiàn)多態(tài),并且提高了性能(只select了一次)
…     polymorphic_on=punion.c.type_,
…     polymorphic_identity=’P')
<Mapper at 0x8605b6c; Product>
>>> mapper(Clothing, clothing_table, inherits=Product,
… polymorphic_identity=’C',
… concrete=True)
<Mapper at 0x84f1bac; Clothing>
>>> mapper(Accessory, accessory_table, inherits=Product,
… polymorphic_identity=’A',
… concrete=True)
<Mapper at 0x858770c; Accessory>

·        >>> session.query(Product).get(’222′)
<Accessory 222>

·        本文主要是講關(guān)于sqlalchemy的擴展

·        擴展其實就是一些外部的插件,比如sqlsoup,associationproxy,declarative,horizontal_shard等等

·        1 declarative

·        假如想要數(shù)據(jù)映射,以前的做法是:

·        from sqlalchemy importcreate_engine

·        from sqlalchemy import Column,MetaData, Table

·        from sqlalchemy import Integer,String, ForeignKey

·        from sqlalchemy.orm importmapper, sessionmaker

·        class User(object): #簡單類

·            def __init__(self, name, fullname,password):

·                self.name = name

·                self.fullname = fullname

·                self.password = password

·            def __repr__(self):

·                return "<User('%s','%s','%s')>" % (self.name, self.fullname, self.password)

·        metadata = MetaData()

·        users_table = Table('users',metadata,

·        Column('user_id', Integer,primary_key=True),

·        Column('name', String),

·        Column('fullname', String),

·        Column('password', String)

·        )

·        email_table = Table('email',metadata,

·        Column('email_id', Integer,primary_key=True),

·        Column('email_address',String),

·        Column('user_id', Integer,ForeignKey('users.user_id'))

·        )

·        metadata.create_all(engine)

·        mapper(User, users_table) #映射

·        但是我們可以該換風(fēng)格,可以用這樣的方法:

·        from sqlalchemy import Column,Integer, String, ForeignKey

·        from sqlalchemy importcreate_engine

·        from sqlalchemy.ext.declarativeimport declarative_base

·        from sqlalchemy.orm importbackref, mapper, relation, sessionmaker

·        Base = declarative_base()

·        class User(Base):

·            __tablename__ = "users" #設(shè)定接收映射的表名

·            id = Column(Integer, primary_key=True) #將表結(jié)構(gòu)寫到類里面

·            name = Column(String)

·            fullname = Column(String)

·            password = Column(String)

·            def __init__(self, name, fullname,password):

·                self.name = name

·                self.fullname = fullname

·                self.password = password

·            def __repr__(self):

·                return "<User('%s','%s','%s')>" % (self.name, self.fullname, self.password)

·        class Address(Base):

·            __tablename__ = "addresses"

·            id = Column(Integer, primary_key=True)

·            email_address = Column(String,nullable=False)

·            user_id = Column(Integer,ForeignKey('users.id'))

·            user = relation(User,backref=backref('addresses', order_by=id)) #創(chuàng)建雙向關(guān)系,標(biāo)識以user的id為連接,也就是說:Address到User是多對一,User到Address是一對多

·            def __init__(self, email_address):

·                self.email_address = email_address

·            def __repr__(self):

·                return"<Address('%s')>" % self.email_address

·        engine =create_engine("sqlite:///tutorial.db", echo=True)

·        users_table = User.__table__ #獲取User表對象句柄

·        metadata = Base.metadata #獲取metadata句柄

·        metadata.create_all(engine)

·        下面具體說:

·        engine =create_engine(‘sqlite://’) #創(chuàng)建引擎
Base.metadata.create_all(engine) #常見表
Base.metadata.bind = create_engine(‘sqlite://’) #綁定
Base = declarative_base(bind=create_engine(‘sqlite://’)) #綁定引擎
mymetadata = MetaData()
Base = declarative_base(metadata=mymetadata) #設(shè)定元數(shù)據(jù)設(shè)定簡單關(guān)系:
class User(Base):
__tablename__ = ‘users’id = Column(Integer, primary_key=True)
name = Column(String(50))
addresses = relationship(“Address”, backref=”user”) #relationship其實就是relation的全稱

·        classAddress(Base):
__tablename__ = ‘a(chǎn)ddresses’

·        id =Column(Integer, primary_key=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey(‘users.id’))
設(shè)定多對多關(guān)系:
keywords = Table(
‘keywords’, Base.metadata,
Column(‘a(chǎn)uthor_id’, Integer, ForeignKey(‘a(chǎn)uthors.id’)),
Column(‘keyword_id’, Integer, ForeignKey(‘keywords.id’))
)
class Author(Base):
__tablename__ = ‘a(chǎn)uthors’
id = Column(Integer, primary_key=True)
keywords = relationship(“Keyword”, secondary=keywords)
定義SQL表達式:
class MyClass(Base):
__tablename__ = ‘sometable’
__table_args__ = {‘mysql_engine’:'InnoDB’} #名字,映射類,元數(shù)據(jù)之外的指定需要使用__table_args__

·        或者:
class MyClass(Base):
__tablename__ = ‘sometable’
__table_args__ = (
ForeignKeyConstraint(['id'], ['remote_table.id']), #元組方式
UniqueConstraint(‘foo’),
)

·        或者:
class MyClass(Base):
__tablename__ = ‘sometable’
__table_args__ = (
ForeignKeyConstraint(['id'], ['remote_table.id']),
UniqueConstraint(‘foo’),
{‘a(chǎn)utoload’:True} #最后的參數(shù)可以用字典 想想*argsand **kwargs
)
使用混合式:
class MyClass(Base):
__table__ = Table(‘my_table’, Base.metadata, #在__table__里指定表結(jié)構(gòu)
Column(‘id’, Integer, primary_key=True),
Column(‘name’, String(50))
)
2 sqlsoup(在sqlalchemy0.8版本后他變成了一個獨立的項目,http://pypi./pypi/sqlsoup,

·        而我使用gentoo提供的0.7.8版本,以下的程序import部分可能不適用更高版本,而需要import sqlsoup)

·        sqlsoup提供一個方便的訪問數(shù)據(jù)庫的接方式,而無需創(chuàng)建類,映射數(shù)據(jù)庫

·        還是看例子的對比:

·        用以前的方式創(chuàng)建一個數(shù)據(jù)庫并且插入一些數(shù)據(jù):

·        >>> fromsqlalchemy import *
>>> engine = create_engine(‘sqlite:///dongwm.db’)
>>> metadata = MetaData(engine)
>>> product_table = Table(
…     ‘product’, metadata,
…     Column(‘sku’, String(20), primary_key=True),
…     Column(‘msrp’, Numeric))
>>> store_table = Table(
…     ‘store’, metadata,
…     Column(‘id’, Integer, primary_key=True),
…     Column(‘name’, Unicode(255)))
>>> product_price_table = Table(
…     ‘product_price’, metadata,
… Column(‘sku2′, None, ForeignKey(‘product.sku’), primary_key=True),
… Column(‘store_id’, None, ForeignKey(‘store.id’), primary_key=True),
…     Column(‘price’, Numeric, default=0))
>>> metadata.create_all()
>>> stmt = product_table.insert()
>>> stmt.execute([dict(sku="123", msrp=12.34),
...              dict(sku="456", msrp=22.12),
...              dict(sku="789", msrp=41.44)])
<sqlalchemy.engine.base.ResultProxy object at 0x84fbdcc>
>>> stmt = store_table.insert()
>>> stmt.execute([dict(name="Main Store"),
...              dict(name="Secondary Store")])
<sqlalchemy.engine.base.ResultProxy object at 0x850068c>
>>> stmt = product_price_table.insert()
>>> stmt.execute([dict(store_id=1, sku="123"),
...              dict(store_id=1, sku2="456"),
...              dict(store_id=1, sku2="789"),
...              dict(store_id=2, sku2="123"),
...              dict(store_id=2, sku2="456"),
...              dict(store_id=2, sku2="789")])
<sqlalchemy.engine.base.ResultProxy object at 0x85008cc>
創(chuàng)建插入完畢,然后我們用sqlsoup連接操作:

·        >>> fromsqlalchemy.ext.sqlsoup import SqlSoup
>>> db = SqlSoup(‘sqlite:///dongwm.db’)  #連接一個存在的數(shù)據(jù)庫
>>> print db.product.all() #打印結(jié)果
[MappedProduct(sku=u'123',msrp=Decimal('12.3400000000')),MappedProduct(sku=u'456',msrp=Decimal('22.1200000000')),MappedProduct(sku=u'789',msrp=Decimal('41.4400000000'))]
>>> print db.product.get(’123′) #是不是比session.query(Product)簡單呢?
MappedProduct(sku=u’123′,msrp=Decimal(’12.3400000000′))

·        注:假如想創(chuàng)建一個數(shù)據(jù)庫: db = SqlSoup(‘sqlite:///:memory:’)

·        >>>newprod = db.product.insert(sku=’111′, msrp=22.44) #沒有使用數(shù)據(jù)映射的插入
>>> db.flush()
>>> db.clear() #調(diào)用底層,清除所有session實例,它是session.expunge_all的別名
>>> db.product.all()
[MappedProduct(sku=u'123',msrp=Decimal('12.3400000000')),MappedProduct(sku=u'456',msrp=Decimal('22.1200000000')),MappedProduct(sku=u'789',msrp=Decimal('41.4400000000')),MappedProduct(sku=u'111',msrp=Decimal('22.4400000000'))] #新條目已經(jīng)存在了
#MappedProduct使用__getattr__將無法識別的屬性和訪問方法轉(zhuǎn)發(fā)到它的query屬性,它還提供了一些數(shù)據(jù)處理功能用于更新

·        >>> fromsqlalchemy import or_, and_, desc
>>> where = or_(db.product.sku==’123′, db.product.sku==’111′)
>>> db.product.filter(where).order_by(desc(db.product.msrp)).all() #這樣使用多條件過濾,降序排練
[MappedProduct(sku='111',msrp=22.44), MappedProduct(sku=u'123',msrp=Decimal('12.3400000000'))]

·        >>> join1= db.join(db.product, db.product_price, isouter=True) #關(guān)聯(lián)2個表, isouter=True確保LEFT OUTER(還沒理解)
>>> join1.all()
[MappedJoin(sku=u'123',msrp=Decimal('12.3400000000'),sku2=u'123',store_id=1,price=Decimal('0E-10')), #這個字段包含了2個表的相應(yīng)字段
MappedJoin(sku=u'123',msrp=Decimal('12.3400000000'),sku2=u'123',store_id=2,price=Decimal('0E-10')),MappedJoin(sku=u'456',msrp=Decimal('22.1200000000'),sku2=u'456',store_id=1,price=Decimal('0E-10')),MappedJoin(sku=u'456',msrp=Decimal('22.1200000000'),sku2=u'456',store_id=2,price=Decimal('0E-10')),MappedJoin(sku=u'789',msrp=Decimal('41.4400000000'),sku2=u'789',store_id=1,price=Decimal('0E-10')),MappedJoin(sku=u'789',msrp=Decimal('41.4400000000'),sku2=u'789',store_id=2,price=Decimal('0E-10')),MappedJoin(sku=u'111',msrp=Decimal('22.4400000000'),sku2=None,store_id=None,price=None)]
>>> join2 = db.join(join1, db.store, isouter=True) #將store表也關(guān)聯(lián)進來(因為也有一個外鍵),就是關(guān)聯(lián)三個表
>>> join2.all()
[MappedJoin(sku=u'123',msrp=Decimal('12.3400000000'),sku2=u'123',store_id=1,price=Decimal('0E-10'),id=1,name=u'MainStore'),MappedJoin(sku=u'123',msrp=Decimal('12.3400000000'),sku2=u'123',store_id=2,price=Decimal('0E-10'),id=2,name=u'SecondaryStore'), MappedJoin(sku=u'456',msrp=Decimal('22.1200000000'),sku2=u'456',store_id=1,price=Decimal('0E-10'),id=1,name=u'MainStore'),MappedJoin(sku=u'456',msrp=Decimal('22.1200000000'),sku2=u'456',store_id=2,price=Decimal('0E-10'),id=2,name=u'SecondaryStore'), MappedJoin(sku=u'789',msrp=Decimal('41.4400000000'),sku2=u'789',store_id=1,price=Decimal('0E-10'),id=1,name=u'MainStore'),MappedJoin(sku=u'789',msrp=Decimal('41.4400000000'),sku2=u'789',store_id=2,price=Decimal('0E-10'),id=2,name=u'SecondaryStore'), MappedJoin(sku=u'111',msrp=Decimal('22.4400000000'),sku2=None,store_id=None,price=None,id=None,name=None)]
>>> join3 = db.with_labels(join1) #根據(jù)原籍標(biāo)記,比如sku會說出:product_sku,告訴你它來著product表,但是指定了jion1,就不會標(biāo)識關(guān)于store的表
>>> join3.first()
MappedJoin(product_sku=u’123′,product_msrp=Decimal(’12.3400000000′),product_price_sku2=u’123′,product_price_store_id=1,product_price_price=Decimal(’0E-10′))
>>> db.with_labels(join2).first()
MappedJoin(product_sku=u’123′,product_msrp=Decimal(’12.3400000000′),product_price_sku2=u’123′,product_price_store_id=1,product_price_price=Decimal(’0E-10′),store_id=1,store_name=u’MainStore’)
>>> labelled_product = db.with_labels(db.product)
>>> join4 = db.join(labelled_product, db.product_price, isouter=True)
>>> join4.first()
MappedJoin(product_sku=u’123′,product_msrp=Decimal(’12.3400000000′),sku2=u’123′,store_id=1,price=Decimal(’0E-10′))

·        >>>db.clear()
>>> join5 = db.join(db.product, db.product_price)
>>> s = select([db.product._table,
...     func.avg(join5.c.price).label('avg_price')], #添加一個字段計算產(chǎn)品(product)的price平均值,字段名為avg_price
…     from_obj=[join5._table],
…     group_by=[join5.c.sku])
>>> s = s.alias(‘products_with_avg_price’) #它是fromsqlalchemy import alias; a = alias(self, name=name)的簡寫
>>> products_with_avg_price = db.map(s, primary_key=[join5.c.sku]) #因為沒有映射到表或者join，需要指定如何找到主鍵
>>> products_with_avg_price.all()
[MappedJoin(sku=u'123',msrp=Decimal('12.3400000000'),avg_price=0.0),MappedJoin(sku=u'456',msrp=Decimal('22.1200000000'),avg_price=0.0),MappedJoin(sku=u'789',msrp=Decimal('41.4400000000'),avg_price=0.0)]
>>> db.product_price.first().price = 50.00
>>> db.flush()
>>> products_with_avg_price.all()
[MappedJoin(sku=u'123',msrp=Decimal('12.3400000000'),avg_price=0.0),MappedJoin(sku=u'456',msrp=Decimal('22.1200000000'),avg_price=0.0),MappedJoin(sku=u'789',msrp=Decimal('41.4400000000'),avg_price=0.0)]
>>> db.products_with_avg_price = products_with_avg_price #保存映射到db,方便重用
>>> msrp=select([db.product.c.msrp],
…     db.product.sku==db.product_price.sku2) #獲取sku和sku2相等時候msrp的值
>>> db.product_price.update(  #更新數(shù)據(jù)
…     values=dict(price=msrp),synchronize_session=False) #設(shè)置price這個字段值為上面對應(yīng)的msrp

·        6
>>> db.product_price.all()
[MappedProduct_price(sku2=u'123',store_id=1,price=Decimal('12.3400000000')),MappedProduct_price(sku2=u'456',store_id=1,price=Decimal('22.1200000000')),MappedProduct_price(sku2=u'789',store_id=1,price=Decimal('41.4400000000')),MappedProduct_price(sku2=u'123',store_id=2,price=Decimal('12.3400000000')),MappedProduct_price(sku2=u'456',store_id=2,price=Decimal('22.1200000000')),MappedProduct_price(sku2=u'789',store_id=2,price=Decimal('41.4400000000'))]

·        3 associationproxy

·        associationproxy用于創(chuàng)建一個讀/寫整個關(guān)系的目標(biāo)屬性

·        看一個例子就懂了:

·        >>>user_table = Table(
…     ‘user’, metadata,
…     Column(‘id’, Integer, primary_key=True),
…     Column(‘user_name’, String(255), unique=True),
…     Column(‘password’, String(255)))
>>> brand_table = Table(
…     ‘brand’, metadata,
…     Column(‘id’, Integer, primary_key=True),
…     Column(‘name’, String(255)))
>>> sales_rep_table = Table(
…     ‘sales_rep’, metadata,
… Column(‘brand_id’, None, ForeignKey(‘brand.id’), primary_key=True),
… Column(‘user_id’, None, ForeignKey(‘user.id’), primary_key=True),
…     Column(‘commission_pct’, Integer, default=0))
>>> class User(object): pass
…
>>> class Brand(object): pass
…
>>> class SalesRep(object): pass
…
>>> mapper(User, user_table, properties=dict(
…     sales_rep=relation(SalesRep, backref=’user’,uselist=False)))
<Mapper at 0x87472ec; User>
>>> mapper(Brand, brand_table, properties=dict(
…     sales_reps=relation(SalesRep, backref=’brand’)))
<Mapper at 0x874770c; Brand>
>>> mapper(SalesRep, sales_rep_table)
<Mapper at 0x874768c; SalesRep>

·        ORM完成,但是假如我們想要brand(品牌)類對象的一個所有SalesReps for Brand(品牌的銷售代表)的User列表屬性,可以這樣:

·        classBrand(object):
@property
def users(self):
return [ sr.user for sr in self.sales_reps ]

·        但是不方便增加刪除,而使用association_proxy:

·        >>> fromsqlalchemy.ext.associationproxy import association_proxy
>>> class Brand(object):
…        users=association_proxy(‘sales_reps’, ‘user’)
…

·        或者:

·        mapper(Brand,brand_table, properties=dict(
sales_reps=relation(SalesRep, backref=’brand’)))
Brand.users=association_proxy(‘sales_reps’, ‘user’)#優(yōu)點是維持了域?qū)ο?/p>

·        我們需要修改類,增加屬性:

·        classUser(object):
def __init__(self, user_name=None, password=None):
self.user_name=user_name
self.password=password

·        classBrand(object):
def __init__(self, name=None):
self.name = name

·        classSalesRep(object):
def __init__(self, user=None, brand=None, commission_pct=0):
self.user = user
self.brand = brand
self.commission_pct=commission_pct

·        看下面的效果:

·        >>> b =Brand(‘Cool Clothing’)
>>> session.add(b)
>>> u = User(‘rick’, ‘foo’)
>>> session.add(u)
>>> session.flush()
2012-07-20 12:22:33,191 INFO sqlalchemy.engine.base.Engine INSERT INTO user(user_name, password) VALUES (?, ?)
2012-07-20 12:22:33,191 INFO sqlalchemy.engine.base.Engine (‘rick’, ‘foo’)
2012-07-20 12:22:33,191 INFO sqlalchemy.engine.base.Engine INSERT INTO brand(name) VALUES (?)
2012-07-20 12:22:33,191 INFO sqlalchemy.engine.base.Engine (‘Cool Clothing’,)
>>> b.users
2012-07-20 12:22:42,135 INFO sqlalchemy.engine.base.Engine SELECTsales_rep.brand_id AS sales_rep_brand_id, sales_rep.user_id ASsales_rep_user_id, sales_rep.commission_pct AS sales_rep_commission_pct
FROM sales_rep
WHERE ? = sales_rep.brand_id
2012-07-20 12:22:42,135 INFO sqlalchemy.engine.base.Engine (2,)
[]
>>> b.users.append(u) #自動創(chuàng)建一個單一的位置參數(shù)調(diào)用其中介(SalesRep)對象
2012-07-20 12:22:46,782 INFO sqlalchemy.engine.base.Engine SELECTsales_rep.brand_id AS sales_rep_brand_id, sales_rep.user_id ASsales_rep_user_id, sales_rep.commission_pct AS sales_rep_commission_pct
FROM sales_rep
WHERE ? = sales_rep.user_id
2012-07-20 12:22:46,782 INFO sqlalchemy.engine.base.Engine (2,)
>>> b.users
[<__main__.User object at 0x87d7b6c>]
>>> b.sales_reps
[<__main__.SalesRep object at 0x87d7c4c>]
>>> b.sales_reps[0].commission_pct
0
>>> session.flush()
2012-07-20 12:23:14,215 INFO sqlalchemy.engine.base.Engine INSERT INTOsales_rep (brand_id, user_id, commission_pct) VALUES (?, ?, ?)
2012-07-20 12:23:14,215 INFO sqlalchemy.engine.base.Engine (2, 2, 0)

·        更復(fù)雜的想法給銷售人員一個10%的提成:

·        Brand.users=association_proxy(
‘sales_reps’, ‘user’,
creator=lambda u:SalesRep(user=u, commission_pct=10))

·        假設(shè)我們想要的品牌屬性是一個附帶User和傭金commission_pct的字典:

·        fromsqlalchemy.orm.collections import attribute_mapped_collection
>>> from sqlalchemy.orm.collections import attribute_mapped_collection
>>> reps_by_user_class=attribute_mapped_collection(‘user’)
>>> clear_mappers()
>>> mapper(Brand, brand_table, properties=dict(
…     sales_reps_by_user=relation(
…         SalesRep, backref=’brand’,
…        collection_class=reps_by_user_class)))
<Mapper at 0x862c5ec; Brand>

·        >>>Brand.commissions=association_proxy(
…     ‘sales_reps_by_user’, ‘commission_pct’,
…     creator=lambda key,value: SalesRep(user=key,commission_pct=value))
>>> mapper(User, user_table, properties=dict(
…     sales_rep=relation(SalesRep, backref=’user’,uselist=False)))
<Mapper at 0x8764b2c; User>
>>> mapper(SalesRep, sales_rep_table)
<Mapper at 0x87bb4cc; SalesRep>
>>> b = session.query(Brand).get(1)
>>> u = session.query(User).get(1)
>>> b.commissions[u] = 20
>>> session.bind.echo = False
>>> session.flush()
>>> b = session.query(Brand).get(1)
>>> u = session.query(User).get(1)
>>> u.user_name
u’dongwm’
>>> print b.commissions[u]
20
>>> print b.sales_reps_by_user[u] #代理和原來的關(guān)系是自動同步的
<__main__.SalesRep object at 0x87e3dcc>
>>> print b.sales_reps_by_user[u].commission_pct
20