[sqlite3] Use TypeAlias for isolation_level (#14460)

This commit is contained in:
Semyon Moroz
2025-07-26 14:26:01 +00:00
committed by GitHub
parent 8317409c18
commit 1bd302042d
2 changed files with 23 additions and 20 deletions
+8 -7
View File
@@ -16,8 +16,9 @@ from sqlite3 import (
ProgrammingError as ProgrammingError,
Row as Row,
Warning as Warning,
_IsolationLevel,
)
from typing import Any, Final, Literal, TypeVar, overload
from typing import Any, Final, TypeVar, overload
from typing_extensions import TypeAlias
if sys.version_info >= (3, 11):
@@ -225,7 +226,7 @@ if sys.version_info >= (3, 12):
database: StrOrBytesPath,
timeout: float = 5.0,
detect_types: int = 0,
isolation_level: Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None = "DEFERRED",
isolation_level: _IsolationLevel = "DEFERRED",
check_same_thread: bool = True,
cached_statements: int = 128,
uri: bool = False,
@@ -237,7 +238,7 @@ if sys.version_info >= (3, 12):
database: StrOrBytesPath,
timeout: float,
detect_types: int,
isolation_level: Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None,
isolation_level: _IsolationLevel,
check_same_thread: bool,
factory: type[_ConnectionT],
cached_statements: int = 128,
@@ -250,7 +251,7 @@ if sys.version_info >= (3, 12):
database: StrOrBytesPath,
timeout: float = 5.0,
detect_types: int = 0,
isolation_level: Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None = "DEFERRED",
isolation_level: _IsolationLevel = "DEFERRED",
check_same_thread: bool = True,
*,
factory: type[_ConnectionT],
@@ -265,7 +266,7 @@ else:
database: StrOrBytesPath,
timeout: float = 5.0,
detect_types: int = 0,
isolation_level: Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None = "DEFERRED",
isolation_level: _IsolationLevel = "DEFERRED",
check_same_thread: bool = True,
cached_statements: int = 128,
uri: bool = False,
@@ -275,7 +276,7 @@ else:
database: StrOrBytesPath,
timeout: float,
detect_types: int,
isolation_level: Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None,
isolation_level: _IsolationLevel,
check_same_thread: bool,
factory: type[_ConnectionT],
cached_statements: int = 128,
@@ -286,7 +287,7 @@ else:
database: StrOrBytesPath,
timeout: float = 5.0,
detect_types: int = 0,
isolation_level: Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None = "DEFERRED",
isolation_level: _IsolationLevel = "DEFERRED",
check_same_thread: bool = True,
*,
factory: type[_ConnectionT],
+15 -13
View File
@@ -220,6 +220,8 @@ _SqliteData: TypeAlias = str | ReadableBuffer | int | float | None
_AdaptedInputData: TypeAlias = _SqliteData | Any
# The Mapping must really be a dict, but making it invariant is too annoying.
_Parameters: TypeAlias = SupportsLenAndGetItem[_AdaptedInputData] | Mapping[str, _AdaptedInputData]
# Controls the legacy transaction handling mode of sqlite3.
_IsolationLevel: TypeAlias = Literal["DEFERRED", "EXCLUSIVE", "IMMEDIATE"] | None
class _AnyParamWindowAggregateClass(Protocol):
def step(self, *args: Any) -> object: ...
@@ -285,7 +287,7 @@ class Connection:
def Warning(self) -> type[Warning]: ...
@property
def in_transaction(self) -> bool: ...
isolation_level: str | None # one of '', 'DEFERRED', 'IMMEDIATE' or 'EXCLUSIVE'
isolation_level: _IsolationLevel
@property
def total_changes(self) -> int: ...
if sys.version_info >= (3, 12):
@@ -299,26 +301,26 @@ class Connection:
def __init__(
self,
database: StrOrBytesPath,
timeout: float = ...,
detect_types: int = ...,
isolation_level: str | None = ...,
check_same_thread: bool = ...,
timeout: float = 5.0,
detect_types: int = 0,
isolation_level: _IsolationLevel = "DEFERRED",
check_same_thread: bool = True,
factory: type[Connection] | None = ...,
cached_statements: int = ...,
uri: bool = ...,
cached_statements: int = 128,
uri: bool = False,
autocommit: bool = ...,
) -> None: ...
else:
def __init__(
self,
database: StrOrBytesPath,
timeout: float = ...,
detect_types: int = ...,
isolation_level: str | None = ...,
check_same_thread: bool = ...,
timeout: float = 5.0,
detect_types: int = 0,
isolation_level: _IsolationLevel = "DEFERRED",
check_same_thread: bool = True,
factory: type[Connection] | None = ...,
cached_statements: int = ...,
uri: bool = ...,
cached_statements: int = 128,
uri: bool = False,
) -> None: ...
def close(self) -> None: ...