摘要：在實際項目中，這么做肯定是不行的實際項目中不會使用內存數據庫，這種數據庫一般只是在單元測試中使用。接下來，我們將會了解中單元測試的相關知識。

在上一篇文章，我們介紹了SQLAlchemy的基本概念，也介紹了基本的使用流程。本文我們結合webdemo這個項目來介紹如何在項目中使用SQLAlchemy。另外，我們還會介紹數據庫版本管理的概念和實踐，這也是OpenStack每個項目都需要做的事情。

我們之前在webdemo項目中已經開發了一個user管理的API，可以在這里回顧。當時只是接收了API請求并且打印信息，并沒有實際的進行數據存儲。現在我們就要引入數據庫操作，來完成user管理的API。

在開發數據庫應用前，需要先定義好數據模型。因為本文只是要演示SQLAlchemy的應用，所以我們定義個最簡單的數據模型。user表的定義如下：

id: 主鍵，一般由數據庫的自增類型實現。

user_id: user id，是一個UUID字符串，是OpenStack中最常用來標記資源的方式，全局唯一，并且為該字段建立索引。

name: user的名稱，允許修改，全局唯一，不能為空。

email: user的email，允許修改，可以為空。

OpenStack項目中我見過兩種數據庫的代碼框架分隔，一種是Keystone的風格，它把一組API的API代碼和數據庫代碼都放在同一個目錄下，如下所示：

采用Pecan框架的項目則大多把數據庫相關代碼都放在db目錄下，比如Magnum項目，如下所示：

由于webdemo采用的是Pecan框架，而且把數據庫操作的代碼放到同一個目錄下也會比較清晰，所以我們采用和Magnum項目相同的方式來編寫數據庫相關的代碼，創建webdemo/db目錄，然后把數據庫操作的相關代碼都放在這個目錄下，如下所示：

由于webdemo項目還沒有使用oslo_db庫，所以代碼看起來比較直觀，沒有Magnum項目復雜。接下來，我們就要開始寫數據庫操作的相關代碼，分為兩個步驟：

在db/models.py中定義User類，對應數據庫的user表。

在db/api.py中實現一個Connection類，這個類封裝了所有的數據庫操作接口。我們會在這個類中實現對user表的CRUD等操作。

db/models.py中的代碼如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext import declarative
from sqlalchemy import Index


Base = declarative.declarative_base()


class User(Base):
    """User table"""

    __tablename__ = "user"
    __table_args__ = (
        Index("ix_user_user_id", "user_id"),
    )
    id = Column(Integer, primary_key=True)
    user_id = Column(String(255), nullable=False)
    name = Column(String(64), nullable=False, unique=True)
    email = Column(String(255))

我們按照我們之前定義的數據模型，實現了映射類。

在db/api.py中，我們先定義了一些通用函數，代碼如下：

from sqlalchemy import create_engine
import sqlalchemy.orm
from sqlalchemy.orm import exc

from webdemo.db import models as db_models


_ENGINE = None
_SESSION_MAKER = None


def get_engine():
    global _ENGINE
    if _ENGINE is not None:
        return _ENGINE

    _ENGINE = create_engine("sqlite://")
    db_models.Base.metadata.create_all(_ENGINE)
    return _ENGINE


def get_session_maker(engine):
    global _SESSION_MAKER
    if _SESSION_MAKER is not None:
        return _SESSION_MAKER

    _SESSION_MAKER = sqlalchemy.orm.sessionmaker(bind=engine)
    return _SESSION_MAKER


def get_session():
    engine = get_engine()
    maker = get_session_maker(engine)
    session = maker()

    return session

上面的代碼中，我們定義了三個函數：

get_engine：返回全局唯一的engine，不需要重復分配。

get_session_maker：返回全局唯一的session maker，不需要重復分配。

get_session：每次返回一個新的session，因為一個session不能同時被兩個數據庫客戶端使用。

這三個函數是使用SQLAlchemy中經常會封裝的，所以OpenStack的oslo_db項目就封裝了這些函數，供所有的OpenStack項目使用。

這里需要注意一個地方，在get_engine()中：

    _ENGINE = create_engine("sqlite://")
    db_models.Base.metadata.create_all(_ENGINE)

我們使用了sqlite內存數據庫，并且立刻創建了所有的表。這么做只是為了演示方便。在實際的項目中，create_engine()的數據庫URL參數應該是從配置文件中讀取的，而且也不能在創建engine后就創建所有的表（這樣數據庫的數據都丟了）。要解決在數據庫中建表的問題，就要先了解數據庫版本管理的知識，也就是database migration，我們在下文中會說明。

Connection的實現就簡單得多了，直接看代碼。這里只實現了get_user()和list_users()方法。

class Connection(object):

    def __init__(self):
        pass

    def get_user(self, user_id):
        query = get_session().query(db_models.User).filter_by(user_id=user_id)
        try:
            user = query.one()
        except exc.NoResultFound:
            # TODO(developer): process this situation
            pass

        return user

    def list_users(self):
        session = get_session()
        query = session.query(db_models.User)
        users = query.all()

        return users

    def update_user(self, user):
        pass

    def delete_user(self, user):
        pass

現在我們有了DB API，接下來就是要在Controller中使用它。對于使用Pecan框架的應用來說，我們定義一個Pecan hook，這個hook在每個請求進來的時候實例化一個db的Connection對象，然后在controller代碼中我們可以直接使用這個Connection實例。關于Pecan hook的相關信息，請查看Pecan官方文檔。

首先，我們要實現這個hook，并且加入到app中。hook的實現代碼在webdemo/api/hooks.py中：

from pecan import hooks

from webdemo.db import api as db_api


class DBHook(hooks.PecanHook):
    """Create a db connection instance."""

    def before(self, state):
        state.request.db_conn = db_api.Connection()

然后，修改webdemo/api/app.py中的setup_app()方法：

def setup_app():
    config = get_pecan_config()

    app_hooks = [hooks.DBHook()]
    app_conf = dict(config.app)
    app = pecan.make_app(
        app_conf.pop("root"),
        logging=getattr(config, "logging", {}),
        hooks=app_hooks,
        **app_conf
    )

    return app

現在，我們就可以在controller使用DB API了。我們這里要重新實現API服務(4)實現的GET /v1/users這個接口：

...
class User(wtypes.Base):
    id = int
    user_id = wtypes.text
    name = wtypes.text
    email = wtypes.text


class Users(wtypes.Base):
    users = [User]
...
class UsersController(rest.RestController):

    @pecan.expose()
    def _lookup(self, user_id, *remainder):
        return UserController(user_id), remainder

    @expose.expose(Users)
    def get(self):
        db_conn = request.db_conn     # 獲取DBHook中創建的Connection實例
        users = db_conn.list_users()  # 調用所需的DB API
        users_list = []
        for user in users:
            u = User()
            u.id = user.id
            u.user_id = user.user_id
            u.name = user.name
            u.email = user.email
            users_list.append(u)
        return Users(users=users_list)

    @expose.expose(None, body=User, status_code=201)
    def post(self, user):
        print user

現在，我們就已經完整的實現了這個API，客戶端訪問API時是從數據庫拿數據，而不是返回一個模擬的數據。讀者可以使用API服務(4)中的方法運行測試服務器來測試這個API。注意：由于數據庫操作依賴于SQLAlchemy庫，所以需要把它添加到requirement.txt中：SQLAlchemy<1.1.0,>=0.9.9。

現在我們已經完成了數據庫層的代碼框架搭建，讀者可以大概了解到一個OpenStack項目中是如何進行數據庫操作的。上面的代碼可以到https://github.com/diabloneo/webdemo下載。

上面我們在get_engine()函數中使用了內存數據庫，并且創建了所有的表。在實際項目中，這么做肯定是不行的：

實際項目中不會使用內存數據庫，這種數據庫一般只是在單元測試中使用。

如果每次create_engine都把數據庫的表重新創建一次，那么數據庫中的數據就丟失了，絕對不可容忍。

解決這個問題的辦法也很簡單：不使用內存數據庫，并且在運行項目代碼前先把數據庫中的表都建好。這么做確實是解決了問題，但是看起來有點麻煩：

如果每次都手動寫SQL語句來創建數據庫中的表，會很容易出錯，而且很麻煩。

如果項目修改了數據模型，那么不能簡單的修改建表的SQL語句，因為重新建表會讓數據丟失。我們只能增加新的SQL語句來修改現有的數據庫。

最關鍵的是：我們怎么知道一個正在生產運行的數據庫是要執行那些SQL語句？如果數據庫第一次使用，那么執行全部的語句是正確的；如果數據庫已經在使用，里面有數據，那么我們只能執行那些修改表定義的SQL語句，而不能執行那些重新建表的SQL語句。

為了解決這種問題，就有人發明了數據庫版本管理的概念，也稱為Database Migration。基本原理是：在我們要使用的數據庫中建立一張表，里面保存了數據庫的當前版本，然后我們在代碼中為每個數據庫版本寫好所需的SQL語句。當對一個數據庫執行migration操作時，會執行從當前版本到目標版本之間的所有SQL語句。舉個例子：

在Version 1時，我們在數據庫中建立一個user表。

在Version 2時，我們在數據庫中建立一個project表。

在Version 3時，我們修改user表，增加一個age列。

那么在我們對一個數據庫執行migration操作，數據庫的當前版本Version 1，我們設定的目標版本是Version 3，那么操作就是：建立一個project表，修改user表，增加一個age列，并且把數據庫當前版本設置為Version 3。

數據庫的版本管理是所有大型數據庫項目的需求，每種語言都有自己的解決方案。OpenStack中主要使用SQLAlchemy的兩種解決方案：sqlalchemy-migrate和Alembic。早期的OpenStack項目使用了sqlalchemy-migrate，后來換成了Alembic。做出這個切換的主要原因是Alembic對數據庫版本的設計和管理更靈活，可以支持分支，而sqlalchemy-migrate只能支持直線的版本管理，具體可以看OpenStack的WiKi文檔Alembic。

接下來，我們就在我們的webdemo項目中引入Alembic來進行版本管理。

要使用Alembic，大概需要以下步驟：

安裝Alembic

在項目中創建Alembic的migration環境

修改Alembic配置文件

創建migration腳本

執行遷移動作

看起來步驟很復雜，其實搭建好環境后，新增數據庫版本只需要執行最后兩個步驟。

在webdemo/requirements.txt中加入：alembic>=0.8.0。然后在virtualenv中安裝即可。

一般OpenStack項目中，Alembic的環境都是放在db/sqlalchemy/目錄下，因此，我們先建立目錄webdemo/db/sqlalchemy/，然后在這個目錄下初始化Alembic環境：

(.venv)? ~/programming/python/webdemo git:(master) ? $ cd webdemo/db
(.venv)? ~/programming/python/webdemo/webdemo/db git:(master) ? $ ls
api.py  api.pyc  __init__.py  __init__.pyc  models.py  models.pyc  sqlalchemy
(.venv)? ~/programming/python/webdemo/webdemo/db git:(master) ? $ cd sqlalchemy
(.venv)? ~/programming/python/webdemo/webdemo/db/sqlalchemy git:(master) ? $ ls
(.venv)? ~/programming/python/webdemo/webdemo/db/sqlalchemy git:(master) ? $ alembic init alembic
  Creating directory /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic ... done
  Creating directory /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic/versions ... done
  Generating /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic/script.py.mako ... done
  Generating /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic.ini ... done
  Generating /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic/README ... done
  Generating /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic/env.pyc ... done
  Generating /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic/env.py ... done
  Please edit configuration/connection/logging settings in "/home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic.ini" before proceeding.
(.venv)? ~/programming/python/webdemo/webdemo/db/sqlalchemy git:(master) ? $

現在，我們就在webdemo/db/sqlalchemy/alembic/目錄下建立了一個Alembic migration環境：

webdemo/db/sqlalchemy/alembic.ini文件是Alembic的配置文件，我們現在需要修改文件中的sqlalchemy.url這個配置項，用來指向我們的數據庫。這里，我們使用SQLite數據庫，數據庫文件存放在webdemo項目的根目錄下，名稱是webdemo.db：

# sqlalchemy.url = driver://user:pass@localhost/dbname
sqlalchemy.url = sqlite:///../../../webdemo.db

注意：實際項目中，數據庫的URL信息是從項目配置文件中讀取，然后通過動態的方式傳遞給Alembic的。具體的做法，讀者可以參考Magnum項目的實現：https://github.com/openstack/magnum/blob/master/magnum/db/sqlalchemy/migration.py。

現在，我們可以創建第一個遷移腳本了，我們的第一個數據庫版本就是創建我們的user表：

(.venv)? ~/programming/python/webdemo/webdemo/db/sqlalchemy git:(master) ? $ alembic revision -m "Create user table"
  Generating /home/diabloneo/programming/python/webdemo/webdemo/db/sqlalchemy/alembic/versions/4bafdb464737_create_user_table.py ... done

現在腳本已經幫我們生成好了，不過這個只是一個空的腳本，我們需要自己實現里面的具體操作，補充完整后的腳本如下：

"""Create user table

Revision ID: 4bafdb464737
Revises:
Create Date: 2016-02-21 12:24:46.640894

"""

# revision identifiers, used by Alembic.
revision = "4bafdb464737"
down_revision = None
branch_labels = None
depends_on = None

from alembic import op
import sqlalchemy as sa


def upgrade():
    op.create_table(
        "user",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("user_id", sa.String(255), nullable=False),
        sa.Column("name", sa.String(64), nullable=False, unique=True),
        sa.Column("email", sa.String(255))
    )


def downgrade():
    op.drop_table("user")

其實就是把User類的定義再寫了一遍，使用了Alembic提供的接口來方便的創建和刪除表。

我們需要在webdemo/db/sqlalchemy/目錄下執行遷移操作，可能需要手動指定PYTHONPATH：

(.venv)? ~/programming/python/webdemo/webdemo/db/sqlalchemy git:(master) ? $ PYTHONPATH=../../../ alembic upgrade head
INFO  [alembic.migration] Context impl SQLiteImpl.
INFO  [alembic.migration] Will assume non-transactional DDL.
INFO  [alembic.migration] Running upgrade  -> 4bafdb464737, Create user table

alembic upgrade head會把數據庫升級到最新的版本。這個時候，在webdemo的根目錄下會出現webdemo.db這個文件，可以使用sqlite3命令查看內容：

(.venv)? ~/programming/python/webdemo git:(master) ? $ ls
AUTHORS  build  ChangeLog  dist  LICENSE  README.md  requirements.txt  Session.vim  setup.cfg  setup.py  webdemo  webdemo.db  webdemo.egg-info
(.venv)? ~/programming/python/webdemo git:(master) ? $ sqlite3 webdemo.db
SQLite version 3.8.11.1 2015-07-29 20:00:57
Enter ".help" for usage hints.
sqlite> .tables
alembic_version  user
sqlite> .schema alembic_version
CREATE TABLE alembic_version (
        version_num VARCHAR(32) NOT NULL
);
sqlite> .schema user
CREATE TABLE user (
        id INTEGER NOT NULL,
        user_id VARCHAR(255) NOT NULL,
        name VARCHAR(64) NOT NULL,
        email VARCHAR(255),
        PRIMARY KEY (id),
        UNIQUE (name)
);
sqlite> .header on
sqlite> select * from alembic_version;
version_num
4bafdb464737

現在我們可以把之前使用的內存數據庫換掉，使用我們的文件數據庫，修改get_engine()函數：

def get_engine():
    global _ENGINE
    if _ENGINE is not None:
        return _ENGINE

    _ENGINE = create_engine("sqlite:///webdemo.db")
    return _ENGINE

現在你可以手動往webdemo.db中添加數據，然后測試下API：

? ~/programming/python/webdemo git:(master) ? $ sqlite3 webdemo.db
SQLite version 3.8.11.1 2015-07-29 20:00:57
Enter ".help" for usage hints.
sqlite> .header on
sqlite> select * from user;
sqlite> .schema user
CREATE TABLE user (
        id INTEGER NOT NULL,
        user_id VARCHAR(255) NOT NULL,
        name VARCHAR(64) NOT NULL,
        email VARCHAR(255),
        PRIMARY KEY (id),
        UNIQUE (name)
);
sqlite> insert into user values(1, "user_id", "Alice", "alice@example.com");
sqlite> select * from user;
id|user_id|name|email
1|user_id|Alice|alice@example.com
sqlite> .q
? ~/programming/python/webdemo git:(master) ? $
? ~/programming/python/webdemo git:(master) ? $ curl http://localhost:8080/v1/users
{"users": [{"email": "alice@example.com", "user_id": "user_id", "id": 1, "name": "Alice"}]}% 

現在，我們就已經完成了database migration代碼框架的搭建，可以成功執行了第一個版本的數據庫遷移。OpenStack項目中也是這么來做數據庫遷移的。后續，一旦修改了項目，需要修改數據模型時，只要新增migration腳本即可。這部分代碼也可以在https://github.com/diabloneo/webdemo中看到。

在實際生產環境中，當我們發布了一個項目的新版本后，在上線的時候，都會自動執行數據庫遷移操作，升級數據庫版本到最新的版本。如果線上的數據庫版本已經是最新的，那么這個操作沒有任何影響；如果不是最新的，那么會把數據庫升級到最新的版本。

關于Alembic的更多使用方法，請閱讀官方文檔Alembic。

本文到這邊就結束了，這兩篇文章我們了解OpenStack中數據庫應用開發的基礎知識。接下來，我們將會了解OpenStack中單元測試的相關知識。