2559 - legal api iam changes

.sonarcloud.properties

@@ -15,4 +15,4 @@ sonar.exclusions=coops-ui.deprecated,schemas.deprecated,**/conftest.py,**/migrat
 sonar.sourceEncoding=UTF-8
 
 # Exclusions for copy-paste detection
-#sonar.cpd.exclusions=
+sonar.cpd.exclusions=**/config.py

legal-api/migrations/env.py

@@ -1,12 +1,11 @@
-from __future__ import with_statement
 
 import logging
+import os
 from logging.config import fileConfig
 
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
-
+import sqlalchemy as sa
 from alembic import context
+from flask import current_app
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
@@ -15,24 +14,28 @@
 # Interpret the config file for Python logging.
 # This line sets up loggers basically.
 fileConfig(config.config_file_name)
-logger = logging.getLogger('alembic.env')
+logger = logging.getLogger("alembic.env")
+
+
+def get_engine():
+    try:
+        # this works with Flask-SQLAlchemy<3 and Alchemical
+        return current_app.extensions["migrate"].db.get_engine()
+    except TypeError:
+        # this works with Flask-SQLAlchemy>=3
+        return current_app.extensions["migrate"].db.engine
 
-# add your model's MetaData object here
-# for 'autogenerate' support
-# from myapp import mymodel
-# target_metadata = mymodel.Base.metadata
-from flask import current_app
-config.set_main_option(
-    'sqlalchemy.url', current_app.config.get(
-        'SQLALCHEMY_DATABASE_URI').replace('%', '%%'))
-target_metadata = current_app.extensions['migrate'].db.metadata
 
-# other values from the config, defined by the needs of env.py,
-# can be acquired:
-# my_important_option = config.get_main_option("my_important_option")
-# ... etc.
+def get_engine_url():
+    try:
+        return get_engine().url.render_as_string(hide_password=False).replace("%", "%%")
+    except AttributeError:
+        return str(get_engine().url).replace("%", "%%")
 
 
+config.set_main_option("sqlalchemy.url", get_engine_url())
+target_metadata = current_app.extensions["migrate"].db.metadata
+
 def run_migrations_offline():
     """Run migrations in 'offline' mode.
 
@@ -66,25 +69,27 @@ def run_migrations_online():
     # when there are no changes to the schema
     # reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
     def process_revision_directives(context, revision, directives):
-        if getattr(config.cmd_opts, 'autogenerate', False):
+        if getattr(config.cmd_opts, "autogenerate", False):
             script = directives[0]
             if script.upgrade_ops.is_empty():
                 directives[:] = []
-                logger.info('No changes in schema detected.')
+                logger.info("No changes in schema detected.")
 
-    connectable = engine_from_config(
-        config.get_section(config.config_ini_section),
-        prefix='sqlalchemy.',
-        poolclass=pool.NullPool,
-    )
+    connectable = get_engine()
 
     with connectable.connect() as connection:
+
+        owner_role = os.getenv("DB_OWNER_ROLE")
+        if owner_role:
+            safe_role = owner_role.replace('"', '""')  # Escape any quotes for SQL safety
+            connection.execute(sa.text(f'SET ROLE "{safe_role}"'))
+
         context.configure(
             connection=connection,
             target_metadata=target_metadata,
             compare_type=True,
             process_revision_directives=process_revision_directives,
-            **current_app.extensions['migrate'].configure_args
+            **current_app.extensions["migrate"].configure_args
         )
 
         with context.begin_transaction():