Remove SQLAlchemy 1.x compatibility branches

With SQLAlchemy 2.0 required, types.DOUBLE, types.Double, and
TypeEngine._variant_mapping always exist. Use them directly and drop the
version-dependent test branches. Keep get_double_type as a public helper
that returns types.DOUBLE.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: laughingman7743

pyathena/sqlalchemy/array.py

@@ -222,8 +222,7 @@ def variant(self, type_: TypeEngine[Any]) -> TypeEngine[Any] | None:
         Returns:
             The variant type, or None when the type has no variant for this dialect.
         """
-        # SQLAlchemy 1.x types have no _variant_mapping.
-        return getattr(type_, "_variant_mapping", {}).get(self.dialect.name)
+        return type_._variant_mapping.get(self.dialect.name)
 
     def decorator_impl(self, type_: types.TypeDecorator[Any]) -> TypeEngine[Any]:
         variant = self.variant(type_)

pyathena/sqlalchemy/base.py

@@ -38,7 +38,6 @@
     AthenaMap,
     AthenaStruct,
     AthenaTimestamp,
-    get_double_type,
 )
 from pyathena.sqlalchemy.util import _HashableDict, _split_type_arguments
 from pyathena.util import (
@@ -71,7 +70,7 @@
 ischema_names: dict[str, type[Any]] = {
     "boolean": types.BOOLEAN,
     "float": types.FLOAT,
-    "double": get_double_type(),
+    "double": types.DOUBLE,
     "real": types.FLOAT,
     "tinyint": TINYINT,
     "smallint": types.SMALLINT,

pyathena/sqlalchemy/compiler.py

@@ -45,7 +45,6 @@
     AthenaMap,
     AthenaStruct,
     AthenaTimestamp,
-    get_double_type,
 )
 from pyathena.sqlalchemy.util import _split_type_arguments
 
@@ -608,7 +607,7 @@ def visit_truediv_binary(self, binary, operator, **kw):
         right_type = binary.right.type
 
         if isinstance(left_type, types.Float) or isinstance(right_type, types.Float):
-            division_type = get_double_type()()
+            division_type: TypeEngine[Any] = types.DOUBLE()
             return (
                 self.process(Cast(binary.left, division_type), **kw)
                 + " / "
@@ -629,7 +628,7 @@ def visit_truediv_binary(self, binary, operator, **kw):
             return (
                 self.process(binary.left, **kw)
                 + " / "
-                + self.process(Cast(binary.right, get_double_type()()), **kw)
+                + self.process(Cast(binary.right, types.DOUBLE()), **kw)
             )
 
         return super().visit_truediv_binary(binary, operator, **kw)
@@ -661,7 +660,7 @@ def visit_cast(self, cast: Cast[Any], **kwargs):
             type_clause = "CHAR"
         elif isinstance(type_, (types.LargeBinary, types.BINARY, types.VARBINARY)):
             type_clause = "VARBINARY"
-        elif hasattr(types, "Double") and isinstance(type_, types.Double):
+        elif isinstance(type_, types.Double):
             type_clause = "DOUBLE"
         elif isinstance(type_, (types.FLOAT, types.Float, types.REAL)):
             # https://docs.aws.amazon.com/athena/latest/ug/data-types.html
@@ -755,7 +754,7 @@ def _complex_dml_type(self, type_, *, require_precision=False, timestamp_precisi
             return "VARCHAR"
         if isinstance(type_, (types.LargeBinary, types.BINARY, types.VARBINARY)):
             return "VARBINARY"
-        if isinstance(type_, getattr(types, "Double", get_double_type())):
+        if isinstance(type_, types.Double):
             return "DOUBLE"
         if isinstance(type_, types.Float):
             return "REAL"

pyathena/sqlalchemy/types.py

@@ -37,18 +37,12 @@
 
 
 def get_double_type() -> type[Any]:
-    """Get the appropriate type for DOUBLE based on SQLAlchemy version.
-
-    SQLAlchemy 2.0+ provides a native DOUBLE type, while earlier versions
-    only have FLOAT. This function returns the appropriate type based on
-    what's available.
+    """Get the SQLAlchemy type for Athena DOUBLE.
 
     Returns:
-        types.DOUBLE for SQLAlchemy 2.0+, types.FLOAT for earlier versions.
+        ``types.DOUBLE``.
     """
-    if hasattr(types, "DOUBLE"):
-        return types.DOUBLE
-    return types.FLOAT
+    return types.DOUBLE
 
 
 class AthenaBinary(types.LargeBinary):

tests/pyathena/sqlalchemy/test_base.py

@@ -31,7 +31,6 @@
     AthenaStruct,
     AthenaTimestamp,
     Tinyint,
-    get_double_type,
 )
 from pyathena.util import RetryConfig
 from tests.pyathena.conftest import ENV
@@ -934,30 +933,13 @@ def test_reflect_table_with_schema(self, engine):
     def test_reflect_table_include_columns(self, engine):
         engine, conn = engine
         one_row_complex = Table("one_row_complex", MetaData(schema=ENV.schema))
-        version = float(re.search(r"^([\d]+\.[\d]+)\..+", sqlalchemy.__version__).group(1))
-        if version <= 1.2:
-            engine.dialect.reflecttable(
-                conn, one_row_complex, include_columns=["col_int"], exclude_columns=[]
-            )
-        elif version == 1.3:
-            # https://docs.sqlalchemy.org/en/13/changelog/changelog_13.html#change-64ac776996da1a5c3e3460b4c0f0b257
-            engine.dialect.reflecttable(
-                conn,
-                one_row_complex,
-                include_columns=["col_int"],
-                exclude_columns=[],
-                resolve_fks=True,
-            )
-        else:  # version >= 1.4
-            # https://docs.sqlalchemy.org/en/14/changelog/changelog_14.html#change-0215fae622c01f9409eb1ba2754f4792
-            # https://docs.sqlalchemy.org/en/14/core/reflection.html#sqlalchemy.engine.reflection.Inspector.reflect_table
-            insp = sqlalchemy.inspect(engine)
-            insp.reflect_table(
-                one_row_complex,
-                include_columns=["col_int"],
-                exclude_columns=[],
-                resolve_fks=True,
-            )
+        insp = sqlalchemy.inspect(engine)
+        insp.reflect_table(
+            one_row_complex,
+            include_columns=["col_int"],
+            exclude_columns=[],
+            resolve_fks=True,
+        )
         assert len(one_row_complex.c) == 1
         assert one_row_complex.c.col_int is not None
         pytest.raises(AttributeError, lambda: one_row_complex.c.col_tinyint)
@@ -1357,7 +1339,7 @@ def test_reflect_select(self, engine):
         assert isinstance(one_row_complex.c.col_int.type, types.INTEGER)
         assert isinstance(one_row_complex.c.col_bigint.type, types.BIGINT)
         assert isinstance(one_row_complex.c.col_float.type, types.FLOAT)
-        assert isinstance(one_row_complex.c.col_double.type, get_double_type())
+        assert isinstance(one_row_complex.c.col_double.type, types.DOUBLE)
         assert isinstance(one_row_complex.c.col_string.type, types.String)
         assert isinstance(one_row_complex.c.col_varchar.type, types.VARCHAR)
         assert one_row_complex.c.col_varchar.type.length == 10
@@ -1408,7 +1390,7 @@ def test_get_column_type(self, engine):
         assert isinstance(dialect._get_column_type("int"), types.INTEGER)
         assert isinstance(dialect._get_column_type("bigint"), types.BIGINT)
         assert isinstance(dialect._get_column_type("float"), types.FLOAT)
-        assert isinstance(dialect._get_column_type("double"), get_double_type())
+        assert isinstance(dialect._get_column_type("double"), types.DOUBLE)
         assert isinstance(dialect._get_column_type("real"), types.FLOAT)
         assert isinstance(dialect._get_column_type("string"), types.String)
         assert isinstance(dialect._get_column_type("varchar"), types.VARCHAR)
@@ -3265,9 +3247,7 @@ def test_numeric_type_variants(self, engine):
         assert type(actual.c.col_integer2.type) in [types.INT, types.INTEGER, types.Integer]
         assert type(actual.c.col_bigint.type) in [types.BIGINT, types.BigInteger]
         assert type(actual.c.col_biginteger.type) in [types.BIGINT, types.BigInteger]
-        expected_double_types = [types.FLOAT, types.Float]
-        if hasattr(types, "DOUBLE"):
-            expected_double_types.extend([types.DOUBLE, types.Double, types.DOUBLE_PRECISION])
+        expected_double_types = [types.DOUBLE, types.Double, types.DOUBLE_PRECISION]
         assert type(actual.c.col_double1.type) in expected_double_types
         assert type(actual.c.col_double2.type) in expected_double_types
         assert type(actual.c.col_double_precision.type) in expected_double_types

tests/pyathena/sqlalchemy/test_types.py

@@ -10,9 +10,5 @@
 def test_get_double_type():
     from pyathena.sqlalchemy.base import ischema_names
 
-    result = get_double_type()
-    if hasattr(types, "DOUBLE"):
-        assert result is types.DOUBLE
-    else:
-        assert result is types.FLOAT
-    assert ischema_names["double"] is result
+    assert get_double_type() is types.DOUBLE
+    assert ischema_names["double"] is types.DOUBLE