zoukankan      html  css  js  c++  java
  • sqlalchemy(二)高级用法

    sqlalchemy(二)高级用法

    本文将介绍sqlalchemy的高级用法。

    外键以及relationship

    首先创建数据库,在这里一个user对应多个address,因此需要在address上增加user_id这个外键(一对多)。

    #!/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='test@test.com'), Address(email_address='test1@test1.com')]
    >>> 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 ('test@test.com', 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 ('test1@test1.com', 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 | test@test.com   |       1 |
    |  2 | test1@test1.com |       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=='test@test.com').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 ('test@test.com',)
    [('jack', 'test@test.com')]
    

    在sqlalchemy中提供了Queqy.join()函数,

    >>> session.query(User).join(Address).filter(Address.email_address=='test@test.com').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 ('test@test.com', 1)
    <demo.User object at 0x7f9a74139a10>
    >>> session.query(User).join(Address).filter(Address.email_address=='test@test.com').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 ('test@test.com', 1)
    'jack'
    >>> session.query(User).join(Address).filter(Address.email_address=='test@test.com').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 ('test@test.com', 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")
    
  • 相关阅读:
    性能测试七:jmeter进阶之文件上传下载、定时器
    性能测试六:jmeter进阶之Cookie与header管理器
    导出文本、表格、图像到PDF格式文件中(学习整理)
    数据库(学习整理)----7--Oracle多表查询,三种join连接
    反射(学习整理)----Class类和加载器ClassLoader类的整理
    反射(学习整理)----操作类、属性、方法!
    数据库(学习整理)----6--Oracle如何快速备份和多次备份数表数据
    数据库(学习整理)----5--Oracle常用的组函数
    数据库(学习整理)----4--Oracle数据查询(基础点1)
    Java小例子(学习整理)-----学生管理系统-控制台版
  • 原文地址:https://www.cnblogs.com/coder2012/p/4746941.html
Copyright © 2011-2022 走看看