raoul
/
openlp
forked from openlp/openlp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

migration to Alembic. 

bzr-revno: 2269
This commit is contained in:
Raoul Snyman 2013-07-05 20:42:44 +02:00 committed by Andreas Preikschat
parent 910587e81f cee5132b86
commit 208b208412
7 changed files with 134 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
openlp/core/lib/db.py
View File

@ -38,6 +38,8 @@ from sqlalchemy import Table, MetaData, Column, types, create_engine
from sqlalchemy.exc import SQLAlchemyError, InvalidRequestError, DBAPIError, OperationalError
from sqlalchemy.orm import scoped_session, sessionmaker, mapper
from sqlalchemy.pool import NullPool
from alembic.migration import MigrationContext
from alembic.operations import Operations
from openlp.core.lib import translate, Settings
from openlp.core.lib.ui import critical_error_message_box
@ -65,6 +67,17 @@ def init_db(url, auto_flush=True, auto_commit=False):
return session, metadata
def get_upgrade_op(session):
"""
Create a migration context and an operations object for performing upgrades.
``session``
The SQLAlchemy session object.
"""
context = MigrationContext.configure(session.bind.connect())
return Operations(context)
def upgrade_db(url, upgrade):
"""
Upgrade a database.
@ -82,13 +95,7 @@ def upgrade_db(url, upgrade):
Provides a class for the metadata table.
"""
pass
load_changes = False
tables = []
try:
tables = upgrade.upgrade_setup(metadata)
load_changes = True
except (SQLAlchemyError, DBAPIError):
pass
metadata_table = Table(u'metadata', metadata,
Column(u'key', types.Unicode(64), primary_key=True),
Column(u'value', types.UnicodeText(), default=None)
@ -105,22 +112,22 @@ def upgrade_db(url, upgrade):
if version > upgrade.__version__:
return version, upgrade.__version__
version += 1
if load_changes:
try:
while hasattr(upgrade, u'upgrade_%d' % version):
log.debug(u'Running upgrade_%d', version)
try:
upgrade_func = getattr(upgrade, u'upgrade_%d' % version)
upgrade_func(session, metadata, tables)
upgrade_func(session, metadata)
session.commit()
# Update the version number AFTER a commit so that we are sure the previous transaction happened
version_meta.value = unicode(version)
session.commit()
version += 1
except (SQLAlchemyError, DBAPIError):
log.exception(u'Could not run database upgrade script '
'"upgrade_%s", upgrade process has been halted.', version)
log.exception(u'Could not run database upgrade script "upgrade_%s", upgrade process has been halted.',
version)
break
else:
except (SQLAlchemyError, DBAPIError):
version_meta = Metadata.populate(key=u'version', value=int(upgrade.__version__))
session.commit()
return int(version_meta.value), upgrade.__version__

19
openlp/plugins/bibles/lib/upgrade.py
View File

@ -37,28 +37,13 @@ __version__ = 1
log = logging.getLogger(__name__)
def upgrade_setup(metadata):
"""
Set up the latest revision all tables, with reflection, needed for the
upgrade process. If you want to drop a table, you need to remove it from
here, and add it to your upgrade function.
"""
# Don't define the "metadata" table, as the upgrade mechanism already
# defines it.
tables = {
u'book': Table(u'book', metadata, autoload=True),
u'verse': Table(u'verse', metadata, autoload=True)
}
return tables
def upgrade_1(session, metadata, tables):
def upgrade_1(session, metadata):
"""
Version 1 upgrade.
This upgrade renames a number of keys to a single naming convention.
"""
metadata_table = metadata.tables[u'metadata']
metadata_table = Table(u'metadata', metadata, autoload=True)
# Copy "Version" to "name" ("version" used by upgrade system)
# TODO: Clean up in a subsequent release of OpenLP (like 2.0 final)
session.execute(insert(metadata_table).values(

51
openlp/plugins/songs/lib/upgrade.py
View File

@ -31,31 +31,15 @@ The :mod:`upgrade` module provides a way for the database and schema that is the
backend for the Songs plugin
"""
from sqlalchemy import Column, Table, types
from sqlalchemy.sql.expression import func
from migrate.changeset.constraint import ForeignKeyConstraint
from sqlalchemy import Column, types
from sqlalchemy.sql.expression import func, false, null, text
from openlp.core.lib.db import get_upgrade_op
__version__ = 3
def upgrade_setup(metadata):
"""
Set up the latest revision all tables, with reflection, needed for the
upgrade process. If you want to drop a table, you need to remove it from
here, and add it to your upgrade function.
"""
tables = {
u'authors': Table(u'authors', metadata, autoload=True),
u'media_files': Table(u'media_files', metadata, autoload=True),
u'song_books': Table(u'song_books', metadata, autoload=True),
u'songs': Table(u'songs', metadata, autoload=True),
u'topics': Table(u'topics', metadata, autoload=True),
u'authors_songs': Table(u'authors_songs', metadata, autoload=True),
u'songs_topics': Table(u'songs_topics', metadata, autoload=True)
}
return tables
def upgrade_1(session, metadata, tables):
def upgrade_1(session, metadata):
"""
Version 1 upgrade.
@ -67,30 +51,35 @@ def upgrade_1(session, metadata, tables):
added to the media_files table, and a weight column so that the media
files can be ordered.
"""
Table(u'media_files_songs', metadata, autoload=True).drop(checkfirst=True)
Column(u'song_id', types.Integer(), default=None).create(table=tables[u'media_files'])
Column(u'weight', types.Integer(), default=0).create(table=tables[u'media_files'])
op = get_upgrade_op(session)
op.drop_table(u'media_files_songs')
op.add_column(u'media_files', Column(u'song_id', types.Integer(), server_default=null()))
op.add_column(u'media_files', Column(u'weight', types.Integer(), server_default=text(u'0')))
if metadata.bind.url.get_dialect().name != 'sqlite':
# SQLite doesn't support ALTER TABLE ADD CONSTRAINT
ForeignKeyConstraint([u'song_id'], [u'songs.id'],
table=tables[u'media_files']).create()
op.create_foreign_key(u'fk_media_files_song_id', u'media_files', u'songs', [u'song_id', u'id'])
def upgrade_2(session, metadata, tables):
def upgrade_2(session, metadata):
"""
Version 2 upgrade.
This upgrade adds a create_date and last_modified date to the songs table
"""
Column(u'create_date', types.DateTime(), default=func.now()).create(table=tables[u'songs'])
Column(u'last_modified', types.DateTime(), default=func.now()).create(table=tables[u'songs'])
op = get_upgrade_op(session)
op.add_column(u'songs', Column(u'create_date', types.DateTime(), default=func.now()))
op.add_column(u'songs', Column(u'last_modified', types.DateTime(), default=func.now()))
def upgrade_3(session, metadata, tables):
def upgrade_3(session, metadata):
"""
Version 3 upgrade.
This upgrade adds a temporary song flag to the songs table
"""
Column(u'temporary', types.Boolean(), default=False).create(table=tables[u'songs'])
op = get_upgrade_op(session)
if metadata.bind.url.get_dialect().name == 'sqlite':
op.add_column(u'songs', Column(u'temporary', types.Boolean(create_constraint=False), server_default=false()))
else:
op.add_column(u'songs', Column(u'temporary', types.Boolean(), server_default=false()))

23
openlp/plugins/songusage/lib/upgrade.py
View File

@ -30,30 +30,19 @@
The :mod:`upgrade` module provides a way for the database and schema that is the
backend for the SongsUsage plugin
"""
from openlp.core.lib.db import get_upgrade_op
from sqlalchemy import Column, Table, types
from sqlalchemy import Column, types
__version__ = 1
def upgrade_setup(metadata):
"""
Set up the latest revision all tables, with reflection, needed for the
upgrade process. If you want to drop a table, you need to remove it from
here, and add it to your upgrade function.
"""
tables = {
u'songusage_data': Table(u'songusage_data', metadata, autoload=True)
}
return tables
def upgrade_1(session, metadata, tables):
def upgrade_1(session, metadata):
"""
Version 1 upgrade.
This upgrade adds two new fields to the songusage database
"""
Column(u'plugin_name', types.Unicode(20), default=u'') \
.create(table=tables[u'songusage_data'], populate_default=True)
Column(u'source', types.Unicode(10), default=u'') \
.create(table=tables[u'songusage_data'], populate_default=True)
op = get_upgrade_op(session)
op.add_column(u'songusage_data', Column(u'plugin_name', types.Unicode(20), server_default=u''))
op.add_column(u'songusage_data', Column(u'source', types.Unicode(10), server_default=u''))

2
setup.py
View File

@ -175,6 +175,8 @@ OpenLP (previously openlp.org) is free church presentation software, or lyrics p
zip_safe=False,
install_requires=[
# -*- Extra requirements: -*-
'sqlalchemy',
'alembic'
],
entry_points="""
# -*- Entry points: -*-

84
tests/functional/openlp_core_lib/test_db.py Normal file
View File

@ -0,0 +1,84 @@
"""
Package to test the openlp.core.lib package.
"""
from unittest import TestCase
from mock import MagicMock, patch
from sqlalchemy.pool import NullPool
from sqlalchemy.orm.scoping import ScopedSession
from sqlalchemy import MetaData
from openlp.core.lib.db import init_db, get_upgrade_op
class TestDB(TestCase):
"""
A test case for all the tests for the :mod:`~openlp.core.lib.db` module.
"""
def init_db_calls_correct_functions_test(self):
"""
Test that the init_db function makes the correct function calls
"""
# GIVEN: Mocked out SQLAlchemy calls and return objects, and an in-memory SQLite database URL
with patch(u'openlp.core.lib.db.create_engine') as mocked_create_engine, \
patch(u'openlp.core.lib.db.MetaData') as MockedMetaData, \
patch(u'openlp.core.lib.db.sessionmaker') as mocked_sessionmaker, \
patch(u'openlp.core.lib.db.scoped_session') as mocked_scoped_session:
mocked_engine = MagicMock()
mocked_metadata = MagicMock()
mocked_sessionmaker_object = MagicMock()
mocked_scoped_session_object = MagicMock()
mocked_create_engine.return_value = mocked_engine
MockedMetaData.return_value = mocked_metadata
mocked_sessionmaker.return_value = mocked_sessionmaker_object
mocked_scoped_session.return_value = mocked_scoped_session_object
db_url = u'sqlite://'
# WHEN: We try to initialise the db
session, metadata = init_db(db_url)
# THEN: We should see the correct function calls
mocked_create_engine.assert_called_with(db_url, poolclass=NullPool)
MockedMetaData.assert_called_with(bind=mocked_engine)
mocked_sessionmaker.assert_called_with(autoflush=True, autocommit=False, bind=mocked_engine)
mocked_scoped_session.assert_called_with(mocked_sessionmaker_object)
self.assertIs(session, mocked_scoped_session_object, u'The ``session`` object should be the mock')
self.assertIs(metadata, mocked_metadata, u'The ``metadata`` object should be the mock')
def init_db_defaults_test(self):
"""
Test that initialising an in-memory SQLite database via ``init_db`` uses the defaults
"""
# GIVEN: An in-memory SQLite URL
db_url = u'sqlite://'
# WHEN: The database is initialised through init_db
session, metadata = init_db(db_url)
# THEN: Valid session and metadata objects should be returned
self.assertIsInstance(session, ScopedSession, u'The ``session`` object should be a ``ScopedSession`` instance')
self.assertIsInstance(metadata, MetaData, u'The ``metadata`` object should be a ``MetaData`` instance')
def get_upgrade_op_test(self):
"""
Test that the ``get_upgrade_op`` function creates a MigrationContext and an Operations object
"""
# GIVEN: Mocked out alembic classes and a mocked out SQLAlchemy session object
with patch(u'openlp.core.lib.db.MigrationContext') as MockedMigrationContext, \
patch(u'openlp.core.lib.db.Operations') as MockedOperations:
mocked_context = MagicMock()
mocked_op = MagicMock()
mocked_connection = MagicMock()
MockedMigrationContext.configure.return_value = mocked_context
MockedOperations.return_value = mocked_op
mocked_session = MagicMock()
mocked_session.bind.connect.return_value = mocked_connection
# WHEN: get_upgrade_op is executed with the mocked session object
op = get_upgrade_op(mocked_session)
# THEN: The op object should be mocked_op, and the correction function calls should have been made
self.assertIs(op, mocked_op, u'The return value should be the mocked object')
mocked_session.bind.connect.assert_called_with()
MockedMigrationContext.configure.assert_called_with(mocked_connection)
MockedOperations.assert_called_with(mocked_context)

1
tests/functional/openlp_core_lib/test_lib.py
View File

@ -16,6 +16,7 @@ from openlp.core.lib import str_to_bool, create_thumb, translate, check_director
TEST_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), u'..', u'..', u'resources'))
class TestLib(TestCase):
def str_to_bool_with_bool_test(self):