]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
Python 3.9 compatible typing
authorPablo Estevez <pablo22estevez@gmail.com>
Mon, 13 Jan 2025 16:31:25 +0000 (13:31 -0300)
committerPablo Estevez <pablo22estevez@gmail.com>
Mon, 13 Jan 2025 16:31:25 +0000 (13:31 -0300)
lib/sqlalchemy/engine/cursor.py
lib/sqlalchemy/engine/default.py
lib/sqlalchemy/engine/interfaces.py
lib/sqlalchemy/sql/compiler.py
lib/sqlalchemy/sql/sqltypes.py
lib/sqlalchemy/util/_collections.py

index 11e0bc147cc7945b1f3409f70bc12672b561afa7..5294bdabd4870cf16b801aaf1c58219102ab6f67 100644 (file)
@@ -1382,7 +1382,7 @@ class FullyBufferedCursorFetchStrategy(CursorFetchStrategy):
     def __init__(
         self,
         dbapi_cursor: Optional[DBAPICursor],
-        alternate_description: _DBAPICursorDescription | None = None,
+        alternate_description: Optional[_DBAPICursorDescription] = None,
         initial_buffer: Optional[Iterable[Any]] = None,
     ):
         self.alternate_cursor_description = alternate_description
index 3582bbc57fe2f53de3869e284018db4a20732356..25902f9318264cc61d970cc80e87ef070ca92662 100644 (file)
@@ -962,10 +962,10 @@ class DefaultDialect(Dialect):
     def is_disconnect(
         self,
         e: Exception,
-        connection: (
-            pool.PoolProxiedConnection | interfaces.DBAPIConnection | None
-        ),
-        cursor: interfaces.DBAPICursor | None,
+        connection: Union[
+            pool.PoolProxiedConnection, interfaces.DBAPIConnection, None
+        ],
+        cursor: Optional[interfaces.DBAPICursor],
     ) -> bool:
         return False
 
@@ -1685,8 +1685,8 @@ class DefaultExecutionContext(ExecutionContext):
     def _execute_scalar(
         self,
         stmt: str,
-        type_: TypeEngine[Any] | None,
-        parameters: _DBAPISingleExecuteParams | None = None,
+        type_: Optional[TypeEngine[Any]],
+        parameters: Optional[_DBAPISingleExecuteParams] = None,
     ) -> Any:
         """Execute a string statement on the current cursor, returning a
         scalar result.
index 310b8fda0aabbbd07d52a81f197c3ec8aac5c30f..4eb54bd16e950a486512f4d537a6ba5be949a9fd 100644 (file)
@@ -782,7 +782,7 @@ class Dialect(EventTarget):
     max_identifier_length: int
     """The maximum length of identifier names."""
 
-    supports_server_side_cursors: generic_fn_descriptor[bool] | bool
+    supports_server_side_cursors: Union[generic_fn_descriptor[bool], bool]
     """indicates if the dialect supports server side cursors"""
 
     server_side_cursors: bool
index b6069bc2402773f2bb766390eb7e4193084d531c..219996445e70b1bc2105cf141e3f7c0e1aeeaa1a 100644 (file)
@@ -3856,7 +3856,7 @@ class SQLCompiler(Compiled):
             return self.render_literal_value(value, bindparam.type)
 
     def render_literal_value(
-        self, value: str | None, type_: sqltypes.String
+        self, value: Optional[str], type_: sqltypes.String
     ) -> str:
         """Render the value of a bind parameter as a quoted literal.
 
@@ -4619,8 +4619,8 @@ class SQLCompiler(Compiled):
         return None
 
     def get_from_hint_text(
-        self, table: FromClause, text: str | None
-    ) -> str | None:
+        self, table: FromClause, text: Optional[str]
+    ) -> Optional[str]:
         return None
 
     def get_crud_hint_text(self, table, text):
@@ -6131,7 +6131,7 @@ class SQLCompiler(Compiled):
 
         return text
 
-    def update_limit_clause(self, update_stmt: "Update") -> str | None:
+    def update_limit_clause(self, update_stmt: "Update") -> Optional[str]:
         """Provide a hook for MySQL to add LIMIT to the UPDATE"""
         return None
 
@@ -6164,7 +6164,10 @@ class SQLCompiler(Compiled):
         )
 
     def visit_update(
-        self, update_stmt: "Update", visiting_cte: CTE | None = None, **kw: Any
+        self,
+        update_stmt: "Update",
+        visiting_cte: Optional[CTE] = None,
+        **kw: Any,
     ) -> str:
         compile_state = update_stmt._compile_state_factory(  # type: ignore[call-arg] # NOQA: E501
             update_stmt, self, **kw  # type: ignore[arg-type]
@@ -6963,13 +6966,13 @@ class DDLCompiler(Compiled):
     def post_create_table(self, table):
         return ""
 
-    def get_column_default_string(self, column: Column[Any]) -> str | None:
+    def get_column_default_string(self, column: Column[Any]) -> Optional[str]:
         if isinstance(column.server_default, schema.DefaultClause):
             return self.render_default_string(column.server_default.arg)
         else:
             return None
 
-    def render_default_string(self, default: Visitable | str) -> str:
+    def render_default_string(self, default: Union[Visitable, str]) -> str:
         if isinstance(default, str):
             return self.sql_compiler.render_literal_value(
                 default, sqltypes.STRINGTYPE
@@ -7162,7 +7165,7 @@ class GenericTypeCompiler(TypeCompiler):
             }
 
     def visit_DECIMAL(
-        self, type_: "sqltypes.DECIMAL[decimal.Decimal| float]", **kw: Any
+        self, type_: sqltypes.DECIMAL[Union[decimal.Decimal, float]], **kw: Any
     ) -> str:
         if type_.precision is None:
             return "DECIMAL"
@@ -7203,9 +7206,9 @@ class GenericTypeCompiler(TypeCompiler):
 
     def _render_string_type(
         self,
-        type_: "sqltypes.String | sqltypes.Uuid[_UUID_RETURN]",
+        type_: Union[sqltypes.String, sqltypes.Uuid[_UUID_RETURN]],
         name: str,
-        length_override: int | None = None,
+        length_override: Optional[int] = None,
     ) -> str:
         text = name
         if length_override:
@@ -7396,7 +7399,7 @@ class IdentifierPreparer:
         self,
         dialect: Dialect,
         initial_quote: str = '"',
-        final_quote: str | None = None,
+        final_quote: Optional[str] = None,
         escape_quote: str = '"',
         quote_case_sensitive_collations: bool = True,
         omit_schema: bool = False,
@@ -7700,7 +7703,7 @@ class IdentifierPreparer:
     @util.preload_module("sqlalchemy.sql.naming")
     def format_constraint(
         self, constraint: Constraint, _alembic_quote: bool = True
-    ) -> str | None:
+    ) -> Optional[str]:
         naming = util.preloaded.sql_naming
 
         if constraint.name is _NONE_NAME:
@@ -7724,7 +7727,7 @@ class IdentifierPreparer:
 
     def truncate_and_render_index_name(
         self, name: str, _alembic_quote: bool = True
-    ) -> str | None:
+    ) -> Optional[str]:
         # calculate these at format time so that ad-hoc changes
         # to dialect.max_identifier_length etc. can be reflected
         # as IdentifierPreparer is long lived
@@ -7738,7 +7741,7 @@ class IdentifierPreparer:
 
     def truncate_and_render_constraint_name(
         self, name: str, _alembic_quote: bool = True
-    ) -> str | None:
+    ) -> Optional[str]:
         # calculate these at format time so that ad-hoc changes
         # to dialect.max_identifier_length etc. can be reflected
         # as IdentifierPreparer is long lived
@@ -7752,7 +7755,7 @@ class IdentifierPreparer:
 
     def _truncate_and_render_maxlen_name(
         self, name: str, max_: int, _alembic_quote: bool
-    ) -> str | None:
+    ) -> Optional[str]:
         if isinstance(name, elements._truncated_label):
             if len(name) > max_:
                 name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
@@ -7770,7 +7773,7 @@ class IdentifierPreparer:
     @overload
     def format_table(
         self,
-        table: "FromClause | None",
+        table: Optional[FromClause],
         use_schema: bool,
         name: str,
     ) -> str: ...
@@ -7785,9 +7788,9 @@ class IdentifierPreparer:
 
     def format_table(
         self,
-        table: "FromClause | None",
+        table: Optional[FromClause],
         use_schema: bool = True,
-        name: str | None = None,
+        name: Optional[str] = None,
     ) -> str:
         """Prepare a quoted table and schema name."""
         if name is None:
@@ -7826,10 +7829,10 @@ class IdentifierPreparer:
         self,
         column: ColumnElement[Any],
         use_table: bool = False,
-        name: str | None = None,
-        table_name: str | None = None,
+        name: Optional[str] = None,
+        table_name: Optional[str] = None,
         use_schema: bool = False,
-        anon_map: Mapping[str, Any] | None = None,
+        anon_map: Optional[Mapping[str, Any]] = None,
     ) -> str:
         """Prepare a quoted column name."""
 
index ef0396bbe5cea66658c4ace458866bc19296934c..116983e8467a1db85288bbd49afdac5f2f0f2869 100644 (file)
@@ -248,7 +248,7 @@ class String(Concatenable, TypeEngine[str]):
 
     def bind_processor(
         self, dialect: Dialect
-    ) -> _BindProcessorType[str] | None:
+    ) -> Optional[_BindProcessorType[str]]:
         return None
 
     def result_processor(
@@ -433,7 +433,10 @@ class NumericCommon(HasExpressionLookup, TypeEngineMixin, Generic[_N]):
         def _type_affinity(
             self,
         ) -> Type[
-            Numeric[decimal.Decimal | float] | Float[decimal.Decimal | float]
+            Union[
+                Numeric[Union[decimal.Decimal, float]],
+                Float[Union[decimal.Decimal, float]],
+            ]
         ]: ...
 
     def __init__(
@@ -661,7 +664,7 @@ class Float(NumericCommon[_N], TypeEngine[_N]):
 
     __visit_name__ = "float"
 
-    scale: int | None = None
+    scale: Optional[int] = None
 
     @overload
     def __init__(
@@ -1333,7 +1336,7 @@ class Enum(String, SchemaType, Emulated, TypeEngine[Union[str, enum.Enum]]):
 
     __visit_name__ = "enum"
 
-    enum_class: None | str | type[enum.Enum]
+    enum_class: Union[None, str, type[enum.Enum]]
 
     def __init__(self, *enums: Union[str, type[enum.Enum]], **kw: Any):
         r"""Construct an enum.
@@ -1488,9 +1491,9 @@ class Enum(String, SchemaType, Emulated, TypeEngine[Union[str, enum.Enum]]):
         """
         self.native_enum = kw.pop("native_enum", True)
         self.create_constraint = kw.pop("create_constraint", False)
-        self.values_callable: (
-            Callable[[type[enum.Enum]], Sequence[str]] | None
-        ) = kw.pop("values_callable", None)
+        self.values_callable: Optional[
+            Callable[[type[enum.Enum]], Sequence[str]]
+        ] = kw.pop("values_callable", None)
         self._sort_key_function = kw.pop("sort_key_function", NO_ARG)
         length_arg = kw.pop("length", NO_ARG)
         self._omit_aliases = kw.pop("omit_aliases", True)
@@ -1540,8 +1543,8 @@ class Enum(String, SchemaType, Emulated, TypeEngine[Union[str, enum.Enum]]):
         )
 
     def _parse_into_values(
-        self, enums: Sequence[str | type[enum.Enum]], kw: Any
-    ) -> tuple[Sequence[str], Sequence[enum.Enum] | Sequence[str]]:
+        self, enums: Sequence[Union[str, type[enum.Enum]]], kw: Any
+    ) -> tuple[Sequence[str], Union[Sequence[enum.Enum], Sequence[str]]]:
         if not enums and "_enums" in kw:
             enums = kw.pop("_enums")
 
@@ -1658,18 +1661,18 @@ class Enum(String, SchemaType, Emulated, TypeEngine[Union[str, enum.Enum]]):
     def _setup_for_values(
         self,
         values: Sequence[str],
-        objects: Sequence[enum.Enum] | Sequence[str],
+        objects: Union[Sequence[enum.Enum], Sequence[str]],
         kw: Any,
     ) -> None:
         self.enums = list(values)
 
-        self._valid_lookup: dict[enum.Enum | str | None, str | None] = dict(
-            zip(reversed(objects), reversed(values))
-        )
+        self._valid_lookup: dict[
+            Union[enum.Enum, str, None], Optional[str]
+        ] = dict(zip(reversed(objects), reversed(values)))
 
-        self._object_lookup: dict[str | None, enum.Enum | str | None] = dict(
-            zip(values, objects)
-        )
+        self._object_lookup: dict[
+            Optional[str], Union[enum.Enum, str, None]
+        ] = dict(zip(values, objects))
 
         self._valid_lookup.update(
             [
@@ -2209,7 +2212,7 @@ class Interval(Emulated, _AbstractInterval, TypeDecorator[dt.timedelta]):
 
     def bind_processor(
         self, dialect: Dialect
-    ) -> "_BindProcessorType[dt.timedelta]":
+    ) -> _BindProcessorType[dt.timedelta]:
         if TYPE_CHECKING:
             assert isinstance(self.impl_instance, DateTime)
         impl_processor = self.impl_instance.bind_processor(dialect)
@@ -3715,7 +3718,7 @@ class Uuid(Emulated, TypeEngine[_UUID_RETURN]):
 
     def bind_processor(
         self, dialect: Dialect
-    ) -> "_BindProcessorType[_UUID_RETURN] | None":
+    ) -> Optional[_BindProcessorType[_UUID_RETURN]]:
         character_based_uuid = (
             not dialect.supports_native_uuid or not self.native_uuid
         )
index 97a38393e98a39839661ac9ee013a7e18617d293..d863dc1e06636df8244623f19e02735845603469 100644 (file)
@@ -431,7 +431,7 @@ def to_column_set(x: Any) -> Set[Any]:
 
 
 def update_copy(
-    d: dict[Any, Any], _new: dict[Any, Any] | None = None, **kw: Any
+    d: dict[Any, Any], _new: Optional[dict[Any, Any]] = None, **kw: Any
 ) -> dict[Any, Any]:
     """Copy the given dict and update with the given values."""