Python - SQLAlchemy

Python 中有很多连接数据库的模块，pymysql、psycopg2、sqlite3 等，但是通常情况下我们都不会单独使用它们，因为程序可能需要支持多个数据库，而且直接通过调用 SQL 的方式去获取数据库中的数据会使得代码维护成本变高，这种情况下可能就要用到 ORM (数据关系映射, Object Relational Mapping)，而 Python 中最著名的 ORM 当属 SQLAlchemy 了

SQLAlchemy 的关系映射提供了 数据库中的表 与 Python 的类、实例之间的映射关系 一般个类对应一张表，一个实例对应一列

SQLAlchemy 基础

首先我们肯定得先安装 SQLAlchemy

$ pip install sqlalchemy

连接数据库

from sqlalchemy import create_engine 

engine = create_engine('sqlite:///:memory:') #通过 create_engine 连接数据库，这里我们使用 sqlite，并存储数据在内存中

# 开启 debug 模式
"""
engine = create_engine('sqlite:///:memeory:', echo=True)
"""

关于连接数据库的 url 可以看官方文档 database-url

Mapping

如果我们需要一张 users 的表，那么我们可以通过一个 User 类来声明它的元数据，它继承自 Base 基类

它有以下几个字段

• username
• email
• location

from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()

"""
SQLAlchemy 通过 Column 这个描述器类来替换数据库字段的访问和赋值过程
"""
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)  # 每一个类都需要主键
    username = Column(String(64), index=True)
    email = Column(String(64), unique=True, index=True)
    location = Column(String(128))

    def __repr__(self):
        """定义 repr 让输出更加直观优雅"""
        return '<User {}>'.format(self.username)

上面的代码用来声明一个数据表的元数据，可以通过 User.__table__ 看看

会返回一个 Table 对象，我们并没有定义 __table__，但是为什么会有这个属性呢？这都是通过 Base 的元类DeclarativeMeta 来完成的

>>> User.__table__
Table('users', MetaData(bind=None), Column('id', Integer(), table=<users>, primary_key=True, nullable=False), Column('username', String(length=64), table=<users>), Column('email', String(length=64), table=<users>), Column('location', String(length=128), table=<users>), schema=None)

创建 Schema

users 表定义了，我们该如何创建呢？可以通过 Metadata 类的 create_all 方法来创建数据表的 schema

>>> Base.metadata.create_all(bind=engine) # 由于指定了 echo 参数，所以打印出创建过程

2017-03-23 20:44:08,121 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2017-03-23 20:44:08,122 INFO sqlalchemy.engine.base.Engine ()
2017-03-23 20:44:08,123 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2017-03-23 20:44:08,123 INFO sqlalchemy.engine.base.Engine ()
2017-03-23 20:44:08,124 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("users")
2017-03-23 20:44:08,124 INFO sqlalchemy.engine.base.Engine ()
2017-03-23 20:44:08,126 INFO sqlalchemy.engine.base.Engine
CREATE TABLE users (
        id INTEGER NOT NULL,
        username VARCHAR(64),
        email VARCHAR(64),
        location VARCHAR(128),
        PRIMARY KEY (id)
)


2017-03-23 20:44:08,126 INFO sqlalchemy.engine.base.Engine ()
2017-03-23 20:44:08,126 INFO sqlalchemy.engine.base.Engine COMMIT
2017-03-23 20:44:08,126 INFO sqlalchemy.engine.base.Engine CREATE INDEX ix_users_username ON users (username)
2017-03-23 20:44:08,126 INFO sqlalchemy.engine.base.Engine ()
2017-03-23 20:44:08,126 INFO sqlalchemy.engine.base.Engine COMMIT
2017-03-23 20:44:08,127 INFO sqlalchemy.engine.base.Engine CREATE UNIQUE INDEX ix_users_email ON users (email)
2017-03-23 20:44:08,127 INFO sqlalchemy.engine.base.Engine ()
2017-03-23 20:44:08,127 INFO sqlalchemy.engine.base.Engine COMMIT

创建 instance

我们已经创建了一张表，怎么给它加字段呢？其实添加一个字段就像新建一个实例一样简单

>>> anyisalin = User(username='AnyISalIn', email='anyisalin@gmail.com', location='ShangHai/China')

>>> anyisalin.__dict__
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState at 0x112aa4438>,
 'email': 'anyisalin@gmail.com',
 'location': 'ShangHai/China',
 'username': 'AnyISalIn'}

CRUD 操作

create

虽然我们添加了一个字段，但是并没有将它保存到数据库中，如果我们需要 存储、查询、更新数据，那么我们就需要定义 Session 类了

from sqlalchemy.orm import sessionmaker

session = sessionmaker(bind=engine)() #就像 Metadata 类一样，需要 bind engine

通过 add 方法将实例临时存储在 Session 对象里面，然后通过 commit 方法提交到数据库，其实 Session 对象的操作就是在进行事务，当然也可以 rollback

>>> session.add(anyisalin)

>>> session.commit() #如果不想提交，并恢复可以调用 session.rollback()
2017-03-23 21:07:47,336 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2017-03-23 21:07:47,338 INFO sqlalchemy.engine.base.Engine INSERT INTO users (username, email, location) VALUES (?, ?, ?)
2017-03-23 21:07:47,338 INFO sqlalchemy.engine.base.Engine ('AnyISalIn', 'anyisalin@gmail.com', 'ShangHai/China')
2017-03-23 21:07:47,338 INFO sqlalchemy.engine.base.Engine COMMIT

read

这里先不介绍读取操作了，后面介绍查询

update

如果想要更新字段的数据，其实也非常简单

>>> anyisalin.username
'AnyISalIn'

>>> anyisalin.username = 'hahahaha' #直接修改实例

>>> session.add(anyisalin)	#重新 add

>>> session.dirty
IdentitySet([<User hahahaha>]) #identiyset中可以看到修改的字段

>>> session.commit() #再次 commit

delete

想要删除字段，调用 delete 方法就可以了

>>> session.delete(anyisalin)

>>> session.commit()

查询

我们先添加一点假的数据以供查询

from faker import Faker # 使用 faker 这个模块来生成一些假数据, pip install faker

faker = Faker()


def fake_data_generate(User, session):
    for _ in range(10):
        user = User(username=faker.name(), email=faker.email(), location=faker.address())
        session.add(user)
    session.commit()

生成假数据

>>> fake_data_generate(User, session)
2017-03-23 21:26:44,761 INFO sqlalchemy.engine.base.Engine INSERT INTO users (username, email, location) VALUES (?, ?, ?)
... ignore output
2017-03-23 21:26:44,766 INFO sqlalchemy.engine.base.Engine COMMIT

基础查询

>>> session.query(User) #指定查询的字段, 如果只指定了表，则会查询所有字段，相当于 select * from table
<sqlalchemy.orm.query.Query at 0x112ceeeb8>

all

返回所有的字段

>>> session.query(User).all()
[<User Sherry Collins>,
 <User Jeffrey Pierce>,
 <User Tamara Sherman>,
 <User Lori James>,
 <User Kendra Hernandez>,
 <User Mitchell Campbell>,
 <User Patricia Maldonado>,
 <User Jennifer Johnson>,
 <User Timothy Williamson>,
 <User Pamela Cochran>]

first

返回第一个字段

>>> session.query(User).first()
<User Sherry Collins>

filter

很多时候需要过滤一些条件，可以通过 filter 方法来实现

>>> session.query(User).filter(User.id == 1)
'<User Sherry Collins>'

count

计数

>>> session.query(User).count()
10

还没写

relationship

如果我们需要将用户的 email 单独存放在一张表里面，并使它们的实例都有一定关联关系，我们就需要定义 relationship 字段了

这是一个简单的例子

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

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()


class Email(Base):
    __tablename__ = 'emails'
    id = Column(Integer, primary_key=True)
    address = Column(String(128))
    user_id = Column(Integer, ForeignKey('users.id'))  # 如果要使用 relationship, 必须有一个引用外键
    user = relationship('User', back_populates='emails')
    """
        back_populates 参数指的是，
        可以通过这个参数来反推,
        也就是说 email.user 可以获取对应用户的实例,
        user.emails 可以获取对应所有邮箱的实例
    """

    def __repr__(self):
        return '<Email {}>'.format(self.address)


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(64))
    emails = relationship('Email', order_by=Email.id, back_populates='user')
    location = Column(String(128))

    def __repr__(self):
        return '<User {}>'.format(self.username)

测试

>>> Base.metadata.create_all(bind=engine)

>>> session = sessionmaker(bind=engine)()

>>> user1 = User(username='anyisalin')

>>> user1.emails
[]

>>> user1.emails.append(Email(address='anyisalin@gmail.com'))

>>> user1.emails.append(Email(address='xilin@vanecloud.com'))

>>> user1.emails
[<Email anyisalin@gmail.com>, <Email xilin@vanecloud.com>]

>>> session.add(user1)

>>> session.commit()

>>> session.query(User).first()
<User anyisalin>

>>> u = session.query(User).first()

>>> u.emails
[<Email xilin@vanecloud.com>, <Email anyisalin@gmail.com>]

>>> u.emails[0].user
<User anyisalin>