Jun 26 MT-2 phase
This commit is contained in:
@@ -158,7 +158,9 @@ Each phase additive; existing tenant-zero traffic keeps working throughout.
|
||||
|
||||
**MT-1 — Tenant resolution + routing. ✅ DONE.** `app/tenancy/` package: `routing.py` (`RoutingSession` subclassing the Flask-SQLAlchemy session), `resolver.py` (Host→tenant via control plane, lazy import), `engine_cache.py` (per-tenant engines), `context.py` (`TenantContext`), `middleware.py` (`init_tenancy` before_request hook + branded unknown-host 404). Edits: `db` init in `app/__init__.py` (+`init_tenancy(app)` call) and `MULTI_TENANT_ENABLED` + pool flags in `config.py`. Gated behind `MULTI_TENANT_ENABLED` (default False) — fully inert until flipped.
|
||||
|
||||
**MT-2 — Per-tenant migration runner.** `flask tenant db upgrade --tenant <id|all>`; record `alembic_head` per tenant.
|
||||
**MT-2 — Per-tenant migration runner. ✅ DONE (with blocker found).** Standalone `migrations_tenant/env.py` (reads URL from config, reuses `migrations/versions` via `version_locations`, no Flask) + `control/tenant_migrate.py` (`upgrade_tenant()`, `current_revision()`, `chain_head()`, CLI: `python -m control.tenant_migrate upgrade|current|heads --tenant <id|slug|all>`). Records `tenants.alembic_head` + a `ProvisioningJob('migrate')` per run. Existing `migrations/env.py` untouched (normal `flask db` still works).
|
||||
|
||||
> **⚠ Blocker found by MT-2:** the live `migrations/versions` chain has **no base** — `phase1_projects_roles.down_revision = '0003_add_user_active'`, which is absent, and no revision has `down_revision = None`. Alembic cannot build the revision map, so `upgrade head` fails against any DB (even a no-op on an up-to-date one). The pre-`phase1` baseline migrations must be restored (or a guarded squashed baseline created) before MT-2 can run against real MySQL DBs and before MT-3 can provision fresh tenants.
|
||||
|
||||
**MT-3 — Provisioning service.** Create DB → create **per-tenant MySQL user + password** + grant scoped to that DB only → upgrade to head → seed first tenant-admin → invite email. Creds encrypted into the `tenants` row. Idempotent, `provisioning_jobs`-logged.
|
||||
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
"""
|
||||
control/tenant_migrate.py
|
||||
-------------------------
|
||||
Per-tenant migration runner (MT-2).
|
||||
|
||||
Runs the EXISTING tenant Alembic chain (migrations/versions) against each
|
||||
tenant's own database via the standalone migrations_tenant/env.py. Records the
|
||||
applied head on tenants.alembic_head and writes a ProvisioningJob row per run.
|
||||
|
||||
The existing migrations are re-run-safe (INFORMATION_SCHEMA guards, Rule 14),
|
||||
so upgrading an already-current tenant (e.g. tenant-zero / LT, whose DB is
|
||||
already at head) is a no-op.
|
||||
|
||||
Usage:
|
||||
python -m control.tenant_migrate upgrade --tenant all
|
||||
python -m control.tenant_migrate upgrade --tenant lts # id or slug
|
||||
python -m control.tenant_migrate current --tenant all
|
||||
python -m control.tenant_migrate heads
|
||||
|
||||
`upgrade_tenant()` is also imported by the MT-3 provisioning service to bring a
|
||||
freshly-created tenant DB up to head.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
from collections import namedtuple
|
||||
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
from alembic.script import ScriptDirectory
|
||||
from alembic.runtime.migration import MigrationContext
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
from control.base import control_session
|
||||
from control.models import Tenant, ProvisioningJob
|
||||
from control.time_utils import now_eastern
|
||||
|
||||
# Repo layout: control/ and migrations/ are siblings at the repo root.
|
||||
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
_DEFAULT_SCRIPT_LOCATION = os.path.join(_REPO_ROOT, 'migrations_tenant')
|
||||
_DEFAULT_VERSION_LOCATIONS = os.path.join(_REPO_ROOT, 'migrations', 'versions')
|
||||
|
||||
# Everything except a deleted tenant should be kept schema-current.
|
||||
MIGRATABLE_STATUSES = ('provisioning', 'active', 'suspended')
|
||||
|
||||
TenantRef = namedtuple('TenantRef', ['id', 'slug', 'db_uri'])
|
||||
|
||||
|
||||
def _make_config(db_uri, script_location=None, version_locations=None):
|
||||
cfg = Config()
|
||||
cfg.set_main_option('script_location', script_location or _DEFAULT_SCRIPT_LOCATION)
|
||||
cfg.set_main_option('version_locations', version_locations or _DEFAULT_VERSION_LOCATIONS)
|
||||
# Escape % so ConfigParser interpolation doesn't choke on URL-encoded passwords.
|
||||
cfg.set_main_option('sqlalchemy.url', db_uri.replace('%', '%%'))
|
||||
return cfg
|
||||
|
||||
|
||||
def chain_head(script_location=None, version_locations=None):
|
||||
"""Return the head revision id of the tenant migration chain."""
|
||||
cfg = _make_config('sqlite://', script_location, version_locations)
|
||||
script = ScriptDirectory.from_config(cfg)
|
||||
heads = script.get_heads()
|
||||
return heads[0] if len(heads) == 1 else ','.join(sorted(heads))
|
||||
|
||||
|
||||
def current_revision(db_uri):
|
||||
"""Return the alembic revision currently stamped on the given database."""
|
||||
engine = create_engine(db_uri, future=True)
|
||||
try:
|
||||
with engine.connect() as conn:
|
||||
return MigrationContext.configure(conn).get_current_revision()
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def upgrade_tenant(tenant, script_location=None, version_locations=None, record_job=True):
|
||||
"""Upgrade one tenant's database to head.
|
||||
|
||||
`tenant` must expose `.id` and `.db_uri` (a control ORM Tenant, a TenantRef,
|
||||
or any object with those attributes). Returns the applied head revision.
|
||||
Records tenants.alembic_head and a ProvisioningJob('migrate') unless
|
||||
record_job is False.
|
||||
"""
|
||||
db_uri = tenant.db_uri
|
||||
cfg = _make_config(db_uri, script_location, version_locations)
|
||||
|
||||
job_id = None
|
||||
if record_job:
|
||||
with control_session() as s:
|
||||
job = ProvisioningJob(tenant_id=tenant.id, action='migrate',
|
||||
status='running', created_at=now_eastern())
|
||||
s.add(job)
|
||||
s.flush()
|
||||
job_id = job.id
|
||||
|
||||
try:
|
||||
# Alembic logs progress to stdout; keep the runner output tidy.
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
command.upgrade(cfg, 'head')
|
||||
applied = current_revision(db_uri)
|
||||
|
||||
with control_session() as s:
|
||||
t = s.get(Tenant, tenant.id)
|
||||
if t is not None:
|
||||
t.alembic_head = applied
|
||||
if job_id is not None:
|
||||
j = s.get(ProvisioningJob, job_id)
|
||||
if j is not None:
|
||||
j.status = 'ok'
|
||||
j.finished_at = now_eastern()
|
||||
j.log = f'upgraded to {applied}'
|
||||
return applied
|
||||
|
||||
except Exception as e:
|
||||
if job_id is not None:
|
||||
with control_session() as s:
|
||||
j = s.get(ProvisioningJob, job_id)
|
||||
if j is not None:
|
||||
j.status = 'failed'
|
||||
j.finished_at = now_eastern()
|
||||
j.log = f'{type(e).__name__}: {e}'
|
||||
raise
|
||||
|
||||
|
||||
# ── CLI ────────────────────────────────────────────────────────────────────
|
||||
|
||||
def _select_tenants(selector):
|
||||
with control_session() as s:
|
||||
q = s.query(Tenant).filter(Tenant.status.in_(MIGRATABLE_STATUSES))
|
||||
if selector != 'all':
|
||||
if selector.isdigit():
|
||||
q = q.filter(Tenant.id == int(selector))
|
||||
else:
|
||||
q = q.filter(Tenant.slug == selector)
|
||||
return [TenantRef(t.id, t.slug, t.db_uri) for t in q.order_by(Tenant.id).all()]
|
||||
|
||||
|
||||
def _cmd_upgrade(args):
|
||||
tenants = _select_tenants(args.tenant)
|
||||
if not tenants:
|
||||
print(f'No matching tenants for --tenant {args.tenant}.')
|
||||
return 0
|
||||
head = chain_head()
|
||||
print(f'Target head: {head}')
|
||||
failures = 0
|
||||
for t in tenants:
|
||||
try:
|
||||
applied = upgrade_tenant(t)
|
||||
print(f' [ok] {t.slug} (id={t.id}) -> {applied}')
|
||||
except Exception as e:
|
||||
failures += 1
|
||||
print(f' [FAIL] {t.slug} (id={t.id}): {type(e).__name__}: {e}')
|
||||
print(f'Done. {len(tenants) - failures}/{len(tenants)} upgraded.')
|
||||
return 1 if failures else 0
|
||||
|
||||
|
||||
def _cmd_current(args):
|
||||
tenants = _select_tenants(args.tenant)
|
||||
if not tenants:
|
||||
print(f'No matching tenants for --tenant {args.tenant}.')
|
||||
return 0
|
||||
head = chain_head()
|
||||
print(f'Chain head: {head}')
|
||||
for t in tenants:
|
||||
try:
|
||||
rev = current_revision(t.db_uri)
|
||||
except Exception as e:
|
||||
rev = f'<error: {type(e).__name__}>'
|
||||
flag = '' if rev == head else ' (BEHIND)'
|
||||
print(f' {t.slug} (id={t.id}): {rev}{flag}')
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_heads(_args):
|
||||
print(chain_head())
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = argparse.ArgumentParser(prog='control.tenant_migrate',
|
||||
description='Per-tenant migration runner')
|
||||
sub = parser.add_subparsers(dest='command', required=True)
|
||||
|
||||
p_up = sub.add_parser('upgrade', help='Upgrade tenant DB(s) to head')
|
||||
p_up.add_argument('--tenant', required=True, help="'all', or a tenant id or slug")
|
||||
p_up.set_defaults(func=_cmd_upgrade)
|
||||
|
||||
p_cur = sub.add_parser('current', help='Show each tenant DB current revision')
|
||||
p_cur.add_argument('--tenant', default='all', help="'all', or a tenant id or slug")
|
||||
p_cur.set_defaults(func=_cmd_current)
|
||||
|
||||
sub.add_parser('heads', help='Show the chain head revision').set_defaults(func=_cmd_heads)
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
sys.exit(args.func(args))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,62 @@
|
||||
"""
|
||||
migrations_tenant/env.py
|
||||
------------------------
|
||||
Standalone Alembic environment for running the EXISTING tenant migration chain
|
||||
(migrations/versions) against an ARBITRARY tenant database.
|
||||
|
||||
Unlike migrations/env.py — which is Flask-Migrate-coupled and always targets
|
||||
the app's default database via current_app — this reads sqlalchemy.url straight
|
||||
from the Alembic config supplied per-tenant by control/tenant_migrate.py, and
|
||||
reuses the same version scripts through version_locations. No Flask import, so
|
||||
it can run for any tenant DB outside an app context.
|
||||
|
||||
The normal `flask db ...` workflow is unaffected; it still uses migrations/env.py.
|
||||
"""
|
||||
|
||||
from logging.config import fileConfig
|
||||
|
||||
from sqlalchemy import create_engine, pool
|
||||
from alembic import context
|
||||
|
||||
config = context.config
|
||||
|
||||
# Only configure logging if an ini file was supplied (the runner builds the
|
||||
# Config programmatically, so this is typically skipped).
|
||||
if config.config_file_name is not None:
|
||||
try:
|
||||
fileConfig(config.config_file_name)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Raw-SQL migrations with INFORMATION_SCHEMA guards — autogenerate is not used,
|
||||
# so no target metadata is required here.
|
||||
target_metadata = None
|
||||
|
||||
|
||||
def run_migrations_offline():
|
||||
context.configure(
|
||||
url=config.get_main_option('sqlalchemy.url'),
|
||||
target_metadata=target_metadata,
|
||||
literal_binds=True,
|
||||
compare_type=False,
|
||||
)
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
def run_migrations_online():
|
||||
url = config.get_main_option('sqlalchemy.url')
|
||||
connectable = create_engine(url, poolclass=pool.NullPool, future=True)
|
||||
try:
|
||||
with connectable.connect() as connection:
|
||||
context.configure(connection=connection, target_metadata=target_metadata)
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
finally:
|
||||
connectable.dispose()
|
||||
|
||||
|
||||
if context.is_offline_mode():
|
||||
run_migrations_offline()
|
||||
else:
|
||||
run_migrations_online()
|
||||
@@ -0,0 +1,22 @@
|
||||
"""${message}
|
||||
|
||||
Revision ID: ${up_revision}
|
||||
Revises: ${down_revision | comma,n}
|
||||
Create Date: ${create_date}
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
${imports if imports else ""}
|
||||
|
||||
revision = ${repr(up_revision)}
|
||||
down_revision = ${repr(down_revision)}
|
||||
branch_labels = ${repr(branch_labels)}
|
||||
depends_on = ${repr(depends_on)}
|
||||
|
||||
|
||||
def upgrade():
|
||||
${upgrades if upgrades else "pass"}
|
||||
|
||||
|
||||
def downgrade():
|
||||
${downgrades if downgrades else "pass"}
|
||||
Reference in New Issue
Block a user