고급 용법

고급 용법

본고는 sqlalchemy의 고급 용법을 소개할 것이다.

외부 키 및 관계

우선 데이터베이스를 만듭니다. 여기에user가 여러 개의address에 대응하기 때문에address에user_를 추가해야 합니다.id 이 키 (1 대 다).

#!/usr/bin/env python
# encoding: utf-8

from sqlalchemy import create_engine
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import ForeignKey
from sqlalchemy.orm import backref
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32))

    addresses = relationship("Address", order_by="Address.id", backref="user")


class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String(32), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    #user = relationship("User", backref=backref('addresses', order_by=id))


engine  = create_engine('mysql://root:root@localhost:3306/test', echo=True)
#Base.metadata.create_all(engine) 

다음은user와address를 호출하여 데이터를 추가합니다.

>>> jack = User(name='jack')
>>> jack.address
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'User' object has no attribute 'address'
>>> jack.addresses
[]
>>> jack.addresses = [Address(email_address='[email protected]'), Address(email_address='[email protected]')]
>>> jack.addresses
[<demo.Address object at 0x7f2536564f90>, <demo.Address object at 0x7f2535dc71d0>]
>>> session.add(jack)
>>> session.commit()
2015-08-19 13:45:36,237 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'
2015-08-19 13:45:36,237 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,238 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()
2015-08-19 13:45:36,238 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8' and `Collation` = 'utf8_bin'
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_1
2015-08-19 13:45:36,239 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_1
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8) COLLATE utf8_bin AS anon_1
2015-08-19 13:45:36,240 INFO sqlalchemy.engine.base.Engine ()
2015-08-19 13:45:36,241 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-08-19 13:45:36,242 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name) VALUES (%s)
2015-08-19 13:45:36,242 INFO sqlalchemy.engine.base.Engine ('jack',)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine INSERT INTO addresses (email_address, user_id) VALUES (%s, %s)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine ('[email protected]', 1L)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine INSERT INTO addresses (email_address, user_id) VALUES (%s, %s)
2015-08-19 13:45:36,243 INFO sqlalchemy.engine.base.Engine ('[email protected]', 1L)
2015-08-19 13:45:36,244 INFO sqlalchemy.engine.base.Engine COMMIT
>>> 

이때 데이터베이스를 보면 방금 삽입한 데이터를 얻을 수 있습니다.

mysql> select * from users;
+----+------+
| id | name |
+----+------+
|  1 | jack |
+----+------+
1 row in set (0.00 sec)

mysql> select * from addresses;
+----+-----------------+---------+
| id | email_address   | user_id |
+----+-----------------+---------+
|  1 | [email protected]   |       1 |
|  2 | [email protected] |       1 |
+----+-----------------+---------+
2 rows in set (0.00 sec)

join 조회

join을 사용하지 않으면 직접 표를 연결해서 조회할 수 있습니다.

>>> session.query(User.name, Address.email_address).filter(User.id==Address.user_id).filter(Address.email_address=='[email protected]').all()
2015-08-19 14:02:02,877 INFO sqlalchemy.engine.base.Engine SELECT users.name AS users_name, addresses.email_address AS addresses_email_address 
FROM users, addresses 
WHERE users.id = addresses.user_id AND addresses.email_address = %s
2015-08-19 14:02:02,878 INFO sqlalchemy.engine.base.Engine ('[email protected]',)
[('jack', '[email protected]')]

sqlalchemy에서 Queqy를 제공합니다.join () 함수,

>>> session.query(User).join(Address).filter(Address.email_address=='[email protected]').first()
2015-08-19 14:06:56,624 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name 
FROM users INNER JOIN addresses ON users.id = addresses.user_id 
WHERE addresses.email_address = %s 
 LIMIT %s
2015-08-19 14:06:56,624 INFO sqlalchemy.engine.base.Engine ('[email protected]', 1)
<demo.User object at 0x7f9a74139a10>
>>> session.query(User).join(Address).filter(Address.email_address=='[email protected]').first().name
2015-08-19 14:07:04,224 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name 
FROM users INNER JOIN addresses ON users.id = addresses.user_id 
WHERE addresses.email_address = %s 
 LIMIT %s
2015-08-19 14:07:04,224 INFO sqlalchemy.engine.base.Engine ('[email protected]', 1)
'jack'
>>> session.query(User).join(Address).filter(Address.email_address=='[email protected]').first().addresses
2015-08-19 14:07:06,534 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name 
FROM users INNER JOIN addresses ON users.id = addresses.user_id 
WHERE addresses.email_address = %s 
 LIMIT %s
2015-08-19 14:07:06,534 INFO sqlalchemy.engine.base.Engine ('[email protected]', 1)
2015-08-19 14:07:06,535 INFO sqlalchemy.engine.base.Engine SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id 
FROM addresses 
WHERE %s = addresses.user_id ORDER BY addresses.id
2015-08-19 14:07:06,535 INFO sqlalchemy.engine.base.Engine (1L,)
[<demo.Address object at 0x7f9a74139350>, <demo.Address object at 0x7f9a741390d0>]
>>> 

주의, 위의 용법의 전제는 외키가 존재하는 상황에서 외키가 없으면 사용할 수 있다.

query.join(Address, User.id==Address.user_id)    # explicit condition
query.join(User.addresses)                       # specify relationship from left to right
query.join(Address, User.addresses)              # same, with explicit target
query.join('addresses')    

테이블의 별칭

>>> from sqlalchemy.orm import aliased
>>> adalias1 = aliased(Address)

하위 쿼리

만약 우리가 이런 조회가 필요하다면,

mysql> SELECT users.*, adr_count.address_count FROM users LEFT OUTER JOIN
    ->     (SELECT user_id, count(*) AS address_count
    ->         FROM addresses GROUP BY user_id) AS adr_count
    ->     ON users.id=adr_count.user_id;
+----+------+---------------+
| id | name | address_count |
+----+------+---------------+
|  1 | jack |             2 |
+----+------+---------------+
1 row in set (0.00 sec)

#　 ， （select user_id ... group_by user_id）
>>> sbq = session.query(Address.user_id, func.count('*').label('address_count')).group_by(Address.user_id).subquery()

#  ， c 
>>> session.query(User.name, sbq.c.address_count).outerjoin(sbq, User.id==sbq.c.user_id).all()
2015-08-19 14:42:53,425 INFO sqlalchemy.engine.base.Engine SELECT users.name AS users_name, anon_1.address_count AS anon_1_address_count
FROM users LEFT OUTER JOIN (SELECT addresses.user_id AS user_id, count(%s) AS address_count
FROM addresses GROUP BY addresses.user_id) AS anon_1 ON users.id = anon_1.user_id
2015-08-19 14:42:53,425 INFO sqlalchemy.engine.base.Engine ('*',)
[('jack', 2L)]
>>>

contains 포함

query.filter(User.addresses.contains(someaddress))

데이터 삭제 delete

>>> session.delete(jack)
>>> session.query(User).filter_by(name='jack').count()
0

외부 키 구성

위의 예에서user - jack을 삭제했지만address의 데이터는 삭제하지 않았습니다.
cascade 필드

addresses = relationship("Address", backref='user',
    cascade="all, delete, delete-orphan")