Fix Flask-SQLAlchemy 3.0 compatibility (hasgeek#1564)

migrations/env.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from logging.config import fileConfig
+from typing import List
 import logging
 
 from alembic import context
@@ -26,14 +27,13 @@
     'sqlalchemy.url',
     str(current_app.extensions['migrate'].db.engines[None].url).replace('%', '%%'),
 )
+bind_names: List[str] = []
 if current_app.config.get('SQLALCHEMY_BINDS') is not None:
     bind_names = list(current_app.config['SQLALCHEMY_BINDS'].keys())
 else:
     get_bind_names = getattr(current_app.extensions['migrate'].db, 'bind_names', None)
     if get_bind_names:
         bind_names = get_bind_names()
-    else:
-        bind_names = []
 for this_bind in bind_names:
     context.config.set_section_option(
         this_bind,
@@ -42,20 +42,22 @@
             '%', '%%'
         ),
     )
-target_metadata = current_app.extensions['migrate'].db.metadata
+target_db = current_app.extensions['migrate'].db
 
-# 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.
+# 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_metadata(bind):
     """Return the metadata for a bind."""
     if bind == '':
         bind = None
+    if hasattr(target_db, 'metadatas'):
+        return target_db.metadatas[bind]
+
+    # legacy, less flexible implementation
     m = MetaData()
-    for t in target_metadata.tables.values():
+    for t in target_db.metadata.tables.values():
         if t.info.get('bind_key') == bind:
             t.tometadata(m)
     return m

migrations/versions/4f805eefa9f4_urls_now_use_string_instead_of_text.py

@@ -0,0 +1,180 @@
+"""URLs now use string instead of text.
+
+Revision ID: 4f805eefa9f4
+Revises: b8a87e6a24f1
+Create Date: 2022-12-22 15:06:54.491988
+
+"""
+
+from typing import Optional, Tuple, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '4f805eefa9f4'
+down_revision: str = 'b8a87e6a24f1'
+branch_labels: Optional[Union[str, Tuple[str, ...]]] = None
+depends_on: Optional[Union[str, Tuple[str, ...]]] = None
+
+
+def upgrade(engine_name='') -> None:
+    """Upgrade all databases."""
+    # Do not modify. Edit `upgrade_` instead
+    globals().get(f'upgrade_{engine_name}', lambda: None)()
+
+
+def downgrade(engine_name='') -> None:
+    """Downgrade all databases."""
+    # Do not modify. Edit `downgrade_` instead
+    globals().get(f'downgrade_{engine_name}', lambda: None)()
+
+
+def upgrade_() -> None:
+    """Upgrade database bind ''."""
+    with op.batch_alter_table('profile', schema=None) as batch_op:
+        batch_op.alter_column(
+            'website',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'logo_url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'banner_image_url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table('project', schema=None) as batch_op:
+        batch_op.alter_column(
+            'website',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'bg_image',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'buy_tickets_url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'banner_video_url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'hasjob_embed_url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table('session', schema=None) as batch_op:
+        batch_op.alter_column(
+            'banner_image_url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table('shortlink', schema=None) as batch_op:
+        batch_op.alter_column(
+            'url',
+            existing_type=sa.UnicodeText(),
+            type_=sa.Unicode(),
+            existing_nullable=False,
+        )
+
+
+def downgrade_() -> None:
+    """Downgrade database bind ''."""
+    with op.batch_alter_table('shortlink', schema=None) as batch_op:
+        batch_op.alter_column(
+            'url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=False,
+        )
+
+    with op.batch_alter_table('session', schema=None) as batch_op:
+        batch_op.alter_column(
+            'banner_image_url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table('project', schema=None) as batch_op:
+        batch_op.alter_column(
+            'hasjob_embed_url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'banner_video_url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'buy_tickets_url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'bg_image',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'website',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table('profile', schema=None) as batch_op:
+        batch_op.alter_column(
+            'banner_image_url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'logo_url',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            'website',
+            existing_type=sa.Unicode(),
+            type_=sa.UnicodeText(),
+            existing_nullable=True,
+        )
+
+
+def upgrade_geoname() -> None:
+    """Upgrade database bind 'geoname'."""
+
+
+def downgrade_geoname() -> None:
+    """Downgrade database bind 'geoname'."""