---
title: SQLAlchemy学习笔记
date: 2018-08-02 16:28:45
tags: [sql, python]
categories: python
---

本文记录SQLAlchemy学习及使用笔记。

<!--more-->

- **如何替换一个已有主键的记录？**

  使用`session.merge()`方法替换`session.add()`，其实就是`SELECT + UPDATE`。

- **插入大量数据**

  - ORM方式

    ```python
    s = Session()
    objects = [
        User(name="u1"),
        User(name="u2"),
        User(name="u3")
    ]
    s.bulk_save_objects(objects)
    ```

    如果插入字典形式的数据，可以使用 [`Session.bulk_insert_mappings()`](http://docs.sqlalchemy.org/en/rel_1_0/orm/session_api.html#sqlalchemy.orm.session.Session.bulk_insert_mappings), 和 [`Session.bulk_update_mappings()`](http://docs.sqlalchemy.org/en/rel_1_0/orm/session_api.html#sqlalchemy.orm.session.Session.bulk_update_mappings) 

    ```python
    s.bulk_insert_mappings(User,
      [dict(name="u1"), dict(name="u2"), dict(name="u3")]
    )
    ```

  - 非ORM方式

    ```python
    session.execute(
        User.__table__.insert(),
        [{'name': `randint(1, 100)`,'age': randint(1, 100)} for i in xrange(10000)]
    )
    session.commit()
    ```

- **获取大量数据**

  - 使用stream方式

    ```python
    def test_core_fetchmany_w_streaming(n):
        """Load Core result rows using fetchmany/streaming."""
    
        with engine.connect() as conn:
            result = conn.execution_options(stream_results=True).\
                execute(Customer.__table__.select().limit(n))
            while True:
                chunk = result.fetchmany(10000)
                if not chunk:
                    break
                for row in chunk:
                    data = row['id'], row['name'], row['description']
    ```

  - 使用yield_per方式

    ```python
    def test_orm_full_objects_chunks(n):
        """Load fully tracked ORM objects a chunk at a time using yield_per()."""
    
        sess = Session(engine)
        for obj in sess.query(Customer).yield_per(1000).enable_eagerloads(False).limit(n):
            pass
    ```