0
Alembic用のカスタムコードを書いています。私のデータベースはプロジェクトの開発環境で常に更新されます。クライアントのすべてのスキーマのためのprototype（ORGSとして動作し、共有データのためのAlembic： 'relation' public.alembic_versionは存在しません。 'version_table_schema`を使用した場合
publicスキーマ- 単一のスキーマのクライアントあたりの「データベース」
- つのスキーマ：プロジェクトは以下を使用してデータベースを必要とします）このとき
、私だけpublicとprototypeスキーマ最新に保ち、複数のクライアント・スキーマについて心配していません。私のenv.pyスクリプトはpublicスキーマではうまくいきますが、prototypeで作業する場合、alembicはpublicのバージョンテーブルを使用しようとしているため、prototypeでは使用できません。
したがって、オプションを使用して、publicスキーマとprototypeスキーマの1つのバージョンテーブルを管理できると考えました。しかし、私がそれを使い始めるとすぐに、私はアップグレードをしようとすると 'の関係 "public.alembic_version"が存在しません。'エラーが発生します。
version_table_schemaを適切なスキーマに設定すると、生成されたリビジョンスクリプトに実際にop.drop_table('alembic_version')という行が含まれている点のみが異なります。この回線は、version_table_schemaが使用されている場合にのみ存在します。
私はちょっとしたことがないと思っています。
以下は参考になるすべての関連ソースファイルです。外部依存関係はappcontextとgetDeclarativeBaseである必要があります。 appcontextは構成のためのものであり、確実に動作しています（データベース接続は正常に動作しています）。 getDeclarativeBaseは、スキーマの宣言ベース（および関連するメタデータ）を動的に取得する方法です。デバッグ出力に基づいて、それも正しく動作しているようです。メタデータ・オブジェクト自体は、構成時に正しいスキーマに関連付けられます。
移行を実行するためのラッパー関数。autoMigratePublic()とautoMigrateOrgProto()は、この場合、移行を行うため
""" A wrapper around Alembic that will assist with database migrations.
Our database system is soemwhat complex to upgrade. There is a public
schema which we use for shared things (such as authentication and job
management). Then there is the org prototype.
There are also individual schemas for each client org. These are trickier.
http://stackoverflow.com/a/35211383/703040
Also useful:
https://github.com/zzzeek/alembic/blob/0e3319bb36f1612e41a8d7da5a48ce1ca33a0b2b/alembic/config.py#L476
"""
import logging
import os
import time
import alembic.config
import alembic.util
CONFIG_ORG_PROTO = os.path.join("migrations", "alembic_orgproto.ini")
CONFIG_PUBLIC = os.path.join("migrations", "alembic_public.ini")
log = logging.getLogger("sys.migrations")
def autoMigratePublic():
try:
freezePublic()
except alembic.util.CommandError:
log.warn("[public] Problem creating migration revision. Skipping as this sometimes just means that we are working with a new DB.")
upgradePublic()
return
def autoMigrateOrgProto():
try:
freezeOrgProto()
except alembic.util.CommandError:
log.warn("[orgproto] Problem creating migration revision. Skipping as this sometimes just means that we are working with a new DB.")
upgradeOrgProto()
return
def freezePublic():
log.info("[public] Checking the database for revisions...")
alembicArgs = [
"--config", CONFIG_PUBLIC,
"--raiseerr",
"revision",
"--autogenerate",
"--message", "autogenerate {0}".format(makeRevisionName()),
]
runAlembic(alembicArgs)
return
def freezeOrgProto():
log.info("[orgproto] Checking the database for revisions...")
alembicArgs = [
"--config", CONFIG_ORG_PROTO,
"--raiseerr",
"revision",
"--autogenerate",
"--message", "autogenerate {0}".format(makeRevisionName()),
]
runAlembic(alembicArgs)
return
def makeRevisionName():
return time.strftime('%Y-%m-%d %H:%M:%S')
def upgradePublic():
log.info("[public] Performing database upgrade...")
alembicArgs = [
"--config", CONFIG_PUBLIC,
"--raiseerr",
"upgrade",
"head",
]
runAlembic(alembicArgs)
return
def upgradeOrgProto():
log.info("[orgproto] Performing database upgrade...")
alembicArgs = [
"--config", CONFIG_ORG_PROTO,
"--raiseerr",
"upgrade",
"head",
]
runAlembic(alembicArgs)
return
def upgradeOrg(schemaName):
log.info("[%s] Performing database upgrade...", schemaName)
alembicArgs = [
"--config", CONFIG_ORG_PROTO,
"--raiseerr",
"upgrade",
"head",
"-x", "schema={0}".format(schemaName),
]
runAlembic(alembicArgs)
return
def runAlembic(args):
return alembic.config.main(args)
クラス問題の方法です。このクラスはenv.pyファイル
import copy
import logging
import os
import re
import traceback
from logging.config import fileConfig
from sqlalchemy import create_engine
import core.context.appcontext
from core.database.declarative import getDeclarativeBase
logging.getLogger("alembic").setLevel(logging.DEBUG)
#==============================================================================
class Migrator(object):
def __init__(self):
from alembic import context
self.context = context
self.config = context.config
self.log = logging.getLogger("sys.migrations")
self.sys = core.context.appcontext.instance()
self.schema = self.config.get_main_option("schema")
if self.context.get_x_argument("schema"):
self.schema = self.context.get_x_argument("schema")
return
def start(self):
import core.database.tables # Make sure the metadata is defined
if self.context.is_offline_mode():
self.log.error("[%s] Can't run migrations offline", self.schema)
return
self.doMigration()
return
def doMigration(self):
targetMetadata = getDeclarativeBase(self.schema).metadata
engine = create_engine(self.sys.getConnectionUrl(), echo=False)
self.log.info("[%s] Engine: %s", self.schema, engine)
self.log.debug("[%s] Metadata: %s", self.schema, targetMetadata)
for t in targetMetadata.sorted_tables:
self.log.debug(" - %s", t)
conn = engine.connect()
try:
self.context.configure(
conn,
version_table_schema=self.schema,
target_metadata=targetMetadata,
process_revision_directives=self.process_revision_directives,
)
with self.context.begin_transaction():
self.context.run_migrations()
finally:
conn.close()
return
def process_revision_directives(self, context, revision, directives):
""" Used to prevent creating empty migrations
See: http://alembic.readthedocs.org/en/latest/cookbook.html#don-t-generate-empty-migrations-with-autogenerate
"""
if self.config.cmd_opts.autogenerate:
script = directives[0]
if script.upgrade_ops.is_empty():
self.log.debug("[%s] Auto-generated migration is empty. No migration file will be created.", self.schema)
directives[:] = []
else:
self.log.info("[%s] Creating new auto-generated migration revision.", self.schema)
return
サンプルenv.pyで参照されています。両方の公共およびプロトタイプのアップグレードは、同じ内容を持つ
from migrations.migrator import Migrator
Migrator().start()
サンプルコンフィグ INI公共およびプロトタイプ移行は
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations/orgproto
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# max length of characters to apply to the
# "slug" field
#truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; this defaults
# to alembic/versions. When using multiple version
# directories, initial revisions must be specified with --version-path
# version_locations = %(here)s/bar %(here)s/bat alembic/versions
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
sqlalchemy.url = <Not Loaded>
schema = orgproto
マコファイルはほぼ正確に同じファイルを使用します私はデフォルトを使用しています。
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
def upgrade():
${upgrades if upgrades else "pass"}
def downgrade():
${downgrades if downgrades else "pass"}