🏷️ fix types

Author: techouse

requirements_dev.txt

@@ -11,7 +11,7 @@ pytest-timeout
 pytimeparse2
 simplejson>=3.19.1
 types-simplejson
-sqlalchemy<2.0.0
+sqlalchemy
 sqlalchemy-utils
 tox
 tqdm>=4.65.0
@@ -20,4 +20,8 @@ packaging>=23.1
 tabulate
 types-tabulate
 Unidecode>=1.3.6
-typing_extensions
+typing_extensions
+types-Pygments
+types-colorama
+types-mock
+types-setuptools

tests/database.py

@@ -12,8 +12,8 @@
 
 
 class Database:
-    engine: Engine = None
-    Session: sessionmaker = None
+    engine: Engine
+    Session: sessionmaker
 
     def __init__(self, database_uri):
         self.Session = sessionmaker()

tests/func/sqlite3_to_mysql_test.py

@@ -17,6 +17,7 @@
 from pytest_mock import MockFixture
 from sqlalchemy import MetaData, Table, create_engine, inspect, select, text
 from sqlalchemy.engine import Connection, CursorResult, Engine, Inspector, Row
+from sqlalchemy.engine.interfaces import ReflectedIndex
 from sqlalchemy.sql import Select
 from sqlalchemy.sql.elements import TextClause
 
@@ -337,21 +338,21 @@ def test_transfer_transfers_all_tables_in_sqlite_file(
 
         """ Test if all the tables have the same indices """
         index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
-        mysql_indices: t.Tuple[t.Dict[str, t.Any], ...] = tuple(
-            {key: index[key] for key in index_keys}
+        mysql_indices: t.Tuple[ReflectedIndex, ...] = tuple(
+            t.cast(ReflectedIndex, {key: index[key] for key in index_keys})
             for index in (chain.from_iterable(mysql_inspect.get_indexes(table_name) for table_name in mysql_tables))
         )
 
         for table_name in sqlite_tables:
-            sqlite_indices: t.List[t.Dict[str, t.Any]] = sqlite_inspect.get_indexes(table_name)
+            sqlite_indices: t.List[ReflectedIndex] = sqlite_inspect.get_indexes(table_name)
             if with_rowid:
                 sqlite_indices.insert(
                     0,
-                    {
-                        "name": "{}_rowid".format(table_name),
-                        "column_names": ["rowid"],
-                        "unique": 1,
-                    },
+                    ReflectedIndex(
+                        name="{}_rowid".format(table_name),
+                        column_names=["rowid"],
+                        unique=True,
+                    ),
                 )
             for sqlite_index in sqlite_indices:
                 sqlite_index["unique"] = bool(sqlite_index["unique"])
@@ -381,13 +382,19 @@ def test_transfer_transfers_all_tables_in_sqlite_file(
                 constraint_type="FOREIGN KEY",
             )
             mysql_fk_result: CursorResult = mysql_cnx.execute(mysql_fk_stmt)
-            mysql_foreign_keys: t.List[t.Dict[str, t.Any]] = [dict(row) for row in mysql_fk_result]
-
-            sqlite_fk_stmt: str = 'PRAGMA foreign_key_list("{table}")'.format(table=table_name)
+            mysql_foreign_keys: t.List[t.Dict[str, t.Any]] = [
+                {
+                    "table": fk["table"],
+                    "from": fk["from"],
+                    "to": fk["to"],
+                }
+                for fk in mysql_fk_result.mappings()
+            ]
+
+            sqlite_fk_stmt: TextClause = text('PRAGMA foreign_key_list("{table}")'.format(table=table_name))
             sqlite_fk_result = sqlite_cnx.execute(sqlite_fk_stmt)
             if sqlite_fk_result.returns_rows:
-                for row in sqlite_fk_result:
-                    fk: t.Dict[str, t.Any] = dict(row)
+                for fk in sqlite_fk_result.mappings():
                     assert {
                         "table": fk["table"],
                         "from": fk["from"],
@@ -398,23 +405,28 @@ def test_transfer_transfers_all_tables_in_sqlite_file(
         sqlite_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
         mysql_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
 
-        meta: MetaData = MetaData(bind=None)
+        meta: MetaData = MetaData()
         for table_name in sqlite_tables:
-            sqlite_table: Table = Table(table_name, meta, autoload=True, autoload_with=sqlite_engine)
-            sqlite_stmt: Select = select([sqlite_table])
-            sqlite_result: t.List[Row] = sqlite_cnx.execute(sqlite_stmt).fetchall()
+            sqlite_table: Table = Table(table_name, meta, autoload_with=sqlite_engine)
+            sqlite_stmt: Select = select(sqlite_table)
+            sqlite_result: t.List[Row[t.Any]] = list(sqlite_cnx.execute(sqlite_stmt).fetchall())
             sqlite_result.sort()
             sqlite_results.append(tuple(tuple(data for data in row) for row in sqlite_result))
 
         for table_name in mysql_tables:
-            mysql_table: Table = Table(table_name, meta, autoload=True, autoload_with=mysql_engine)
-            mysql_stmt: Select = select([mysql_table])
-            mysql_result: t.List[Row] = mysql_cnx.execute(mysql_stmt).fetchall()
+            mysql_table: Table = Table(table_name, meta, autoload_with=mysql_engine)
+            mysql_stmt: Select = select(mysql_table)
+            mysql_result: t.List[Row[t.Any]] = list(mysql_cnx.execute(mysql_stmt).fetchall())
             mysql_result.sort()
             mysql_results.append(tuple(tuple(data for data in row) for row in mysql_result))
 
         assert sqlite_results == mysql_results
 
+        mysql_cnx.close()
+        sqlite_cnx.close()
+        mysql_engine.dispose()
+        sqlite_engine.dispose()
+
     @pytest.mark.transfer
     @pytest.mark.parametrize(
         "chunk, with_rowid, mysql_insert_method, ignore_duplicate_keys",
@@ -518,21 +530,21 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_sqlite_fi
 
         """ Test if all the tables have the same indices """
         index_keys: t.Tuple[str, ...] = ("name", "column_names", "unique")
-        mysql_indices: t.Tuple[t.Dict[str, t.Any], ...] = tuple(
-            {key: index[key] for key in index_keys}
+        mysql_indices: t.Tuple[ReflectedIndex, ...] = tuple(
+            t.cast(ReflectedIndex, {key: index[key] for key in index_keys})
             for index in (chain.from_iterable(mysql_inspect.get_indexes(table_name) for table_name in mysql_tables))
         )
 
         for table_name in random_sqlite_tables:
-            sqlite_indices: t.List[t.Dict[str, t.Any]] = sqlite_inspect.get_indexes(table_name)
+            sqlite_indices: t.List[ReflectedIndex] = sqlite_inspect.get_indexes(table_name)
             if with_rowid:
                 sqlite_indices.insert(
                     0,
-                    {
-                        "name": "{}_rowid".format(table_name),
-                        "column_names": ["rowid"],
-                        "unique": 1,
-                    },
+                    ReflectedIndex(
+                        name="{}_rowid".format(table_name),
+                        column_names=["rowid"],
+                        unique=True,
+                    ),
                 )
             for sqlite_index in sqlite_indices:
                 sqlite_index["unique"] = bool(sqlite_index["unique"])
@@ -544,19 +556,24 @@ def test_transfer_specific_tables_transfers_only_specified_tables_from_sqlite_fi
         sqlite_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
         mysql_results: t.List[t.Tuple[t.Tuple[t.Any, ...], ...]] = []
 
-        meta: MetaData = MetaData(bind=None)
+        meta: MetaData = MetaData()
         for table_name in random_sqlite_tables:
-            sqlite_table: Table = Table(table_name, meta, autoload=True, autoload_with=sqlite_engine)
-            sqlite_stmt: Select = select([sqlite_table])
-            sqlite_result: t.List[Row] = sqlite_cnx.execute(sqlite_stmt).fetchall()
+            sqlite_table: Table = Table(table_name, meta, autoload_with=sqlite_engine)
+            sqlite_stmt: Select = select(sqlite_table)
+            sqlite_result: t.List[Row[t.Any]] = list(sqlite_cnx.execute(sqlite_stmt).fetchall())
             sqlite_result.sort()
             sqlite_results.append(tuple(tuple(data for data in row) for row in sqlite_result))
 
         for table_name in mysql_tables:
-            mysql_table: Table = Table(table_name, meta, autoload=True, autoload_with=mysql_engine)
-            mysql_stmt: Select = select([mysql_table])
-            mysql_result: t.List[Row] = mysql_cnx.execute(mysql_stmt).fetchall()
+            mysql_table: Table = Table(table_name, meta, autoload_with=mysql_engine)
+            mysql_stmt: Select = select(mysql_table)
+            mysql_result: t.List[Row[t.Any]] = list(mysql_cnx.execute(mysql_stmt).fetchall())
             mysql_result.sort()
             mysql_results.append(tuple(tuple(data for data in row) for row in mysql_result))
 
         assert sqlite_results == mysql_results
+
+        mysql_cnx.close()
+        sqlite_cnx.close()
+        mysql_engine.dispose()
+        sqlite_engine.dispose()

tests/models.py

@@ -12,57 +12,53 @@
     TIMESTAMP,
     VARCHAR,
     BigInteger,
-    Boolean,
     Column,
-    Date,
-    DateTime,
+    Dialect,
     ForeignKey,
     Integer,
     SmallInteger,
     String,
     Table,
     Text,
-    Time,
 )
-from sqlalchemy.dialects.sqlite.base import SQLiteDialect
-from sqlalchemy.orm import backref, declarative_base, relationship
-from sqlalchemy.orm.decl_api import DeclarativeMeta
+from sqlalchemy.orm import DeclarativeBase, Mapped, backref, mapped_column, relationship
 from sqlalchemy.sql.functions import current_timestamp
 
 
 class SQLiteNumeric(types.TypeDecorator):
     impl: t.Type[String] = types.String
 
-    def load_dialect_impl(self, dialect: SQLiteDialect) -> t.Any:
+    def load_dialect_impl(self, dialect: Dialect) -> t.Any:
         return dialect.type_descriptor(types.VARCHAR(100))
 
-    def process_bind_param(self, value: t.Any, dialect: SQLiteDialect) -> str:
+    def process_bind_param(self, value: t.Any, dialect: Dialect) -> str:
         return str(value)
 
-    def process_result_value(self, value: t.Any, dialect: SQLiteDialect) -> Decimal:
+    def process_result_value(self, value: t.Any, dialect: Dialect) -> Decimal:
         return Decimal(value)
 
 
 class MyCustomType(types.TypeDecorator):
     impl: t.Type[String] = types.String
 
-    def load_dialect_impl(self, dialect: SQLiteDialect) -> t.Any:
+    def load_dialect_impl(self, dialect: Dialect) -> t.Any:
         return dialect.type_descriptor(types.VARCHAR(self.length))
 
-    def process_bind_param(self, value: t.Any, dialect: SQLiteDialect) -> str:
+    def process_bind_param(self, value: t.Any, dialect: Dialect) -> str:
         return str(value)
 
-    def process_result_value(self, value: t.Any, dialect: SQLiteDialect) -> str:
+    def process_result_value(self, value: t.Any, dialect: Dialect) -> str:
         return str(value)
 
 
-Base: DeclarativeMeta = declarative_base()
+class Base(DeclarativeBase):
+    pass
 
 
 class Author(Base):
     __tablename__ = "authors"
-    id: int = Column(Integer, primary_key=True)
-    name: str = Column(String(128), nullable=False, index=True)
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
 
     def __repr__(self):
         return "<Author(id='{id}', name='{name}')>".format(id=self.id, name=self.name)
@@ -78,9 +74,9 @@ def __repr__(self):
 
 class Image(Base):
     __tablename__ = "images"
-    id: int = Column(Integer, primary_key=True)
-    path: str = Column(String(255), index=True)
-    description: str = Column(String(255), nullable=True)
+    id: Mapped[int] = mapped_column(primary_key=True)
+    path: Mapped[str] = mapped_column(String(255), index=True)
+    description: Mapped[str] = mapped_column(String(255), nullable=True)
 
     def __repr__(self):
         return "<Image(id='{id}', path='{path}')>".format(id=self.id, path=self.path)
@@ -96,8 +92,8 @@ def __repr__(self):
 
 class Tag(Base):
     __tablename__ = "tags"
-    id: int = Column(Integer, primary_key=True)
-    name: str = Column(String(128), nullable=False, index=True)
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
 
     def __repr__(self):
         return "<Tag(id='{id}', name='{name}')>".format(id=self.id, name=self.name)
@@ -115,27 +111,27 @@ class Misc(Base):
     """This model contains all possible MySQL types"""
 
     __tablename__ = "misc"
-    id: int = Column(Integer, primary_key=True)
-    big_integer_field: int = Column(BigInteger, default=0)
-    blob_field: bytes = Column(BLOB, nullable=True, index=True)
-    boolean_field: bool = Column(Boolean, default=False)
-    char_field: str = Column(CHAR(255), nullable=True)
-    date_field: date = Column(Date, nullable=True)
-    date_time_field: datetime = Column(DateTime, nullable=True)
-    decimal_field: Decimal = Column(SQLiteNumeric(10, 2), nullable=True)
-    float_field: Decimal = Column(SQLiteNumeric(12, 4), default=0)
-    integer_field: int = Column(Integer, default=0)
+    id: Mapped[int] = mapped_column(primary_key=True)
+    big_integer_field: Mapped[int] = mapped_column(BigInteger, default=0)
+    blob_field: Mapped[bytes] = mapped_column(BLOB, nullable=True, index=True)
+    boolean_field: Mapped[bool] = mapped_column(default=False)
+    char_field: Mapped[str] = mapped_column(CHAR(255), nullable=True)
+    date_field: Mapped[date] = mapped_column(nullable=True)
+    date_time_field: Mapped[datetime] = mapped_column(nullable=True)
+    decimal_field: Mapped[Decimal] = mapped_column(SQLiteNumeric(10, 2), nullable=True)
+    float_field: Mapped[Decimal] = mapped_column(SQLiteNumeric(12, 4), default=0)
+    integer_field: Mapped[int] = mapped_column(default=0)
     if environ.get("LEGACY_DB", "0") == "0":
-        json_field: t.Dict[str, t.Any] = Column(JSON, nullable=True)
-    numeric_field: Decimal = Column(SQLiteNumeric(12, 4), default=0)
-    real_field: float = Column(REAL(12, 4), default=0)
-    small_integer_field: int = Column(SmallInteger, default=0)
-    string_field: str = Column(String(255), nullable=True)
-    text_field: str = Column(Text, nullable=True)
-    time_field: time = Column(Time, nullable=True)
-    varchar_field: str = Column(VARCHAR(255), nullable=True)
-    timestamp_field: datetime = Column(TIMESTAMP, default=current_timestamp())
-    my_type_field: t.Any = Column(MyCustomType(255), nullable=True)
+        json_field: Mapped[t.Dict[str, t.Any]] = mapped_column(JSON, nullable=True)
+    numeric_field: Mapped[Decimal] = mapped_column(SQLiteNumeric(12, 4), default=0)
+    real_field: Mapped[float] = mapped_column(REAL(12, 4), default=0)
+    small_integer_field: Mapped[int] = mapped_column(SmallInteger, default=0)
+    string_field: Mapped[str] = mapped_column(String(255), nullable=True)
+    text_field: Mapped[str] = mapped_column(Text, nullable=True)
+    time_field: Mapped[time] = mapped_column(nullable=True)
+    varchar_field: Mapped[str] = mapped_column(VARCHAR(255), nullable=True)
+    timestamp_field: Mapped[datetime] = mapped_column(TIMESTAMP, default=current_timestamp())
+    my_type_field: Mapped[t.Any] = mapped_column(MyCustomType(255), nullable=True)
 
 
 article_misc: Table = Table(
@@ -148,9 +144,9 @@ class Misc(Base):
 
 class Media(Base):
     __tablename__ = "media"
-    id: str = Column(CHAR(64), primary_key=True)
-    title: str = Column(String(255), index=True)
-    description: str = Column(String(255), nullable=True)
+    id: Mapped[str] = mapped_column(CHAR(64), primary_key=True)
+    title: Mapped[str] = mapped_column(String(255), index=True)
+    description: Mapped[str] = mapped_column(String(255), nullable=True)
 
     def __repr__(self):
         return "<Media(id='{id}', title='{title}')>".format(id=self.id, title=self.title)
@@ -166,39 +162,39 @@ def __repr__(self):
 
 class Article(Base):
     __tablename__ = "articles"
-    id: int = Column(Integer, primary_key=True)
-    hash: str = Column(String(32), unique=True)
-    slug: str = Column(String(255), index=True)
-    title: str = Column(String(255), index=True)
-    content: str = Column(Text, nullable=True, index=True)
-    status: str = Column(CHAR(1), index=True)
-    published: datetime = Column(DateTime, nullable=True)
+    id: Mapped[int] = mapped_column(primary_key=True)
+    hash: Mapped[str] = mapped_column(String(32), unique=True)
+    slug: Mapped[str] = mapped_column(String(255), index=True)
+    title: Mapped[str] = mapped_column(String(255), index=True)
+    content: Mapped[str] = mapped_column(Text, nullable=True, index=True)
+    status: Mapped[str] = mapped_column(CHAR(1), index=True)
+    published: Mapped[datetime] = mapped_column(nullable=True)
     # relationships
-    authors: t.List[Author] = relationship(
+    authors: Mapped[t.List[Author]] = relationship(
         "Author",
         secondary=article_authors,
         backref=backref("authors", lazy="dynamic"),
         lazy="dynamic",
     )
-    tags: t.List[Tag] = relationship(
+    tags: Mapped[t.List[Tag]] = relationship(
         "Tag",
         secondary=article_tags,
         backref=backref("tags", lazy="dynamic"),
         lazy="dynamic",
     )
-    images: t.List[Image] = relationship(
+    images: Mapped[t.List[Image]] = relationship(
         "Image",
         secondary=article_images,
         backref=backref("images", lazy="dynamic"),
         lazy="dynamic",
     )
-    media: t.List[Media] = relationship(
+    media: Mapped[t.List[Media]] = relationship(
         "Media",
         secondary=article_media,
         backref=backref("media", lazy="dynamic"),
         lazy="dynamic",
     )
-    misc: t.List[Misc] = relationship(
+    misc: Mapped[t.List[Misc]] = relationship(
         "Misc",
         secondary=article_misc,
         backref=backref("misc", lazy="dynamic"),