153 lines
5.1 KiB
Python
153 lines
5.1 KiB
Python
"""Préparation des schémas PostgreSQL par groupe.
|
|
|
|
Cette couche pose l'isolation physique sans encore changer le chemin de lecture
|
|
principal de l'application. Les schémas tenant sont remplis depuis les tables
|
|
publiques et servent de base de migration vers l'exécution par search_path.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import uuid
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
|
|
|
|
TENANT_TABLES = [
|
|
"groupes",
|
|
"membres",
|
|
"postes",
|
|
"postes_modules",
|
|
"config_repartition",
|
|
"destinataires_contributions",
|
|
"litterature",
|
|
"reserves",
|
|
"reunions",
|
|
"presences",
|
|
"pv_reunions",
|
|
"rapports_rsg",
|
|
"candidatures",
|
|
"propositions",
|
|
"affectations",
|
|
"rotations",
|
|
"collectes",
|
|
"depenses",
|
|
"envois_contributions",
|
|
"evenements",
|
|
"invitations",
|
|
"jetons",
|
|
"ventes_jetons",
|
|
"ventes_litterature",
|
|
"operations_bancaires",
|
|
"mouvements_reserves",
|
|
"transactions_comptables",
|
|
"rapports_adoptes",
|
|
"push_subscriptions",
|
|
"sync_log",
|
|
"historique_modifications",
|
|
"journal_actions",
|
|
]
|
|
|
|
CHILD_TABLE_FILTERS = {
|
|
"postes_modules": "poste_id IN (SELECT id FROM public.postes WHERE groupe_id = :gid)",
|
|
"presences": "reunion_id IN (SELECT id FROM public.reunions WHERE groupe_id = :gid)",
|
|
"pv_reunions": "reunion_id IN (SELECT id FROM public.reunions WHERE groupe_id = :gid)",
|
|
}
|
|
|
|
|
|
def tenant_schema_name(groupe_id) -> str:
|
|
valeur = str(groupe_id)
|
|
try:
|
|
valeur = uuid.UUID(valeur).hex
|
|
except ValueError:
|
|
valeur = re.sub(r"[^a-zA-Z0-9_]", "_", valeur).lower()
|
|
return f"grp_{valeur}"
|
|
|
|
|
|
def _quote_ident(name: str) -> str:
|
|
if not re.fullmatch(r"[a-z_][a-z0-9_]*", name):
|
|
raise ValueError("Nom de schéma invalide")
|
|
return f'"{name}"'
|
|
|
|
|
|
async def ensure_tenant_schema_column(conn: AsyncConnection) -> None:
|
|
await conn.execute(text("ALTER TABLE groupes ADD COLUMN IF NOT EXISTS tenant_schema VARCHAR(80) UNIQUE"))
|
|
await conn.execute(text(
|
|
"UPDATE groupes SET tenant_schema = 'grp_' || replace(id::text, '-', '') "
|
|
"WHERE tenant_schema IS NULL"
|
|
))
|
|
|
|
|
|
async def ensure_tenant_schema(conn: AsyncConnection, schema: str) -> None:
|
|
schema_ident = _quote_ident(schema)
|
|
await conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_ident}"))
|
|
for table in TENANT_TABLES:
|
|
await conn.execute(text(
|
|
f"CREATE TABLE IF NOT EXISTS {schema_ident}.{_quote_ident(table)} "
|
|
f"(LIKE public.{_quote_ident(table)} INCLUDING ALL)"
|
|
))
|
|
|
|
|
|
async def recreate_tenant_tables(conn: AsyncConnection, schema: str) -> None:
|
|
schema_ident = _quote_ident(schema)
|
|
await conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_ident}"))
|
|
for table in reversed(TENANT_TABLES):
|
|
await conn.execute(text(f"DROP TABLE IF EXISTS {schema_ident}.{_quote_ident(table)} CASCADE"))
|
|
for table in TENANT_TABLES:
|
|
await conn.execute(text(
|
|
f"CREATE TABLE {schema_ident}.{_quote_ident(table)} "
|
|
f"(LIKE public.{_quote_ident(table)} INCLUDING ALL)"
|
|
))
|
|
|
|
|
|
async def ensure_all_tenant_schemas(conn: AsyncConnection) -> None:
|
|
await ensure_tenant_schema_column(conn)
|
|
result = await conn.execute(text("SELECT tenant_schema FROM groupes WHERE tenant_schema IS NOT NULL"))
|
|
for schema in result.scalars().all():
|
|
await ensure_tenant_schema(conn, schema)
|
|
|
|
|
|
async def ensure_group_tenant_schema(db: AsyncSession, groupe) -> str:
|
|
schema = groupe.tenant_schema or tenant_schema_name(groupe.id)
|
|
groupe.tenant_schema = schema
|
|
await db.flush()
|
|
conn = await db.connection()
|
|
await ensure_tenant_schema(conn, schema)
|
|
return schema
|
|
|
|
|
|
async def refresh_group_tenant_schema(db: AsyncSession, groupe_id) -> dict:
|
|
conn = await db.connection()
|
|
await ensure_tenant_schema_column(conn)
|
|
result = await conn.execute(
|
|
text("SELECT tenant_schema FROM groupes WHERE id = :gid"),
|
|
{"gid": str(groupe_id)},
|
|
)
|
|
schema = result.scalar_one_or_none()
|
|
if not schema:
|
|
schema = tenant_schema_name(groupe_id)
|
|
await conn.execute(
|
|
text("UPDATE groupes SET tenant_schema = :schema WHERE id = :gid"),
|
|
{"schema": schema, "gid": str(groupe_id)},
|
|
)
|
|
await recreate_tenant_tables(conn, schema)
|
|
schema_ident = _quote_ident(schema)
|
|
|
|
params = {"gid": str(groupe_id)}
|
|
for table in TENANT_TABLES:
|
|
table_ident = _quote_ident(table)
|
|
if table == "groupes":
|
|
where_clause = "id = :gid"
|
|
elif table in CHILD_TABLE_FILTERS:
|
|
where_clause = CHILD_TABLE_FILTERS[table]
|
|
else:
|
|
where_clause = "groupe_id = :gid"
|
|
await conn.execute(text(
|
|
f"INSERT INTO {schema_ident}.{table_ident} "
|
|
f"SELECT * FROM public.{table_ident} WHERE {where_clause}"
|
|
), params)
|
|
return {"schema": schema, "tables": len(TENANT_TABLES)}
|
|
|
|
|
|
async def refresh_all_tenant_schemas(db: AsyncSession) -> list[dict]:
|
|
result = await db.execute(text("SELECT id FROM groupes ORDER BY nom"))
|
|
return [await refresh_group_tenant_schema(db, groupe_id) for groupe_id in result.scalars().all()]
|