mypy: Adds some additional stubs and cleans up obsolete overrides

Author: Daverball

.pre-commit-config.yaml

@@ -24,6 +24,7 @@ repos:
           - 'flake8-markupsafe@git+https://github.com/vmagamedov/flake8-markupsafe@b391bd13df9330e01666d304b7f4403d67e5ceba'
           - flake8-noqa==1.3.2
           - flake8-pyi==23.11.0
+          - flake8-type-checking==2.8.0
   - repo: local
     hooks:
       - id: system

pyproject.toml

@@ -48,14 +48,10 @@ mypy_path = "$MYPY_CONFIG_FILE_DIR/src:$MYPY_CONFIG_FILE_DIR/stubs"
 
 [[tool.mypy.overrides]]
 module = [
-    "alembic.*",
-    "magic.*",
     "plaster.*",
     "pyramid.*",
     "pyramid_beaker.*",
     "pyramid_layout.*",
-    "transaction.*",
-    "zope.sqlalchemy.*"
 ]
 ignore_missing_imports = true
 disallow_untyped_defs = false

stubs/transaction/__init__.pyi

@@ -0,0 +1,25 @@
+from collections.abc import Callable, Iterable
+from typing import Protocol
+
+from transaction._manager import Attempt, ThreadTransactionManager, TransactionManager as TransactionManager
+from transaction._transaction import Savepoint as Savepoint, Transaction as Transaction
+from transaction.interfaces import ITransaction, ITransactionManager
+
+
+class _CreateSavepoint(Protocol):
+    def __call__(self, optimistic: bool = False) -> Savepoint: ...
+
+class _CreateAttempts(Protocol):
+    def __call__(self, number: int = 3) -> Iterable[Attempt]: ...
+
+manager: ThreadTransactionManager
+get: Callable[[], Transaction]
+__enter__: Callable[[], Transaction]
+begin: Callable[[], Transaction]
+commit: Callable[[], None]
+abort: Callable[[], None]
+__exit__: Callable[[object, object, object], None]
+doom: Callable[[], None]
+isDoomed: Callable[[], bool]
+savepoint: _CreateSavepoint
+attempts: _CreateAttempts

stubs/transaction/_manager.pyi

@@ -0,0 +1,72 @@
+import threading
+from collections.abc import Callable, Iterator
+from typing import ClassVar, TypeVar, overload
+
+from transaction._transaction import Savepoint, Transaction
+from transaction.interfaces import _Synchronizer, ITransactionManager
+from zope.interface import implementer
+
+_T = TypeVar("_T")
+
+@implementer(ITransactionManager)
+class TransactionManager:
+    explicit: bool
+    def __init__(self, explicit: bool = False) -> None: ...
+    def begin(self) -> Transaction: ...
+    def __enter__(self) -> Transaction: ...
+    def get(self) -> Transaction: ...
+    def free(self, txn: Transaction) -> None: ...
+    def registerSynch(self, synch: _Synchronizer) -> None: ...
+    def unregisterSynch(self, synch: _Synchronizer) -> None: ...
+    def clearSynchs(self) -> None: ...
+    def registeredSynchs(self) -> bool: ...
+    def isDoomed(self) -> bool: ...
+    def doom(self) -> None: ...
+    def commit(self) -> None: ...
+    def abort(self) -> None: ...
+    def __exit__(self, t: object, v: object, tb: object) -> None: ...
+    def savepoint(self, optimistic: bool = False) -> Savepoint: ...
+    def attempts(self, number: int = 3) -> Iterator[Attempt]: ...
+    run_no_func_types: ClassVar[tuple[type[object], ...]]
+    @overload
+    def run(self, func: Callable[[], _T], tries: int = 3) -> _T: ...
+    @overload
+    def run(self, func: None = None, tries: int = 3) -> Callable[[Callable[[], _T]], _T]: ...
+    @overload
+    def run(self, tries: int, /) -> Callable[[Callable[[], _T]], _T]: ...
+
+@implementer(ITransactionManager)
+class ThreadTransactionManager(threading.local):
+    manager: TransactionManager
+    def __init__(self) -> None: ...
+    @property
+    def explicit(self) -> bool: ...
+    @explicit.setter
+    def explicit(self, v: bool) -> None: ...
+    def begin(self) -> Transaction: ...
+    def get(self) -> Transaction: ...
+    def __enter__(self) -> Transaction: ...
+    def commit(self) -> None: ...
+    def abort(self) -> None: ...
+    def __exit__(self, t: object, v: object, tb: object) -> None: ...
+    def doom(self) -> None: ...
+    def isDoomed(self) -> bool: ...
+    def savepoint(self, optimistic: bool = False) -> Savepoint: ...
+    def registerSynch(self, synch: _Synchronizer) -> None: ...
+    def unregisterSynch(self, synch: _Synchronizer) -> None: ...
+    def clearSynchs(self) -> None: ...
+    def registeredSynchs(self) -> bool: ...
+    def attempts(self, number: int = 3) -> Iterator[Attempt]: ...
+    @overload
+    def run(self, func: Callable[[], _T], tries: int = 3) -> _T: ...
+    @overload
+    def run(self, func: None = None, tries: int = 3) -> Callable[[Callable[[], _T]], _T]: ...
+    @overload
+    def run(self, tries: int, /) -> Callable[[Callable[[], _T]], _T]: ...
+
+class Attempt:
+    success: bool
+    manager: TransactionManager
+    def __init__(self, manager: TransactionManager) -> None: ...
+    def __enter__(self) -> Transaction: ...
+    def __exit__(self, t: object, v: object, tb: object) -> bool: ...

stubs/transaction/_transaction.pyi

@@ -0,0 +1,99 @@
+from collections.abc import Callable, Iterator, Sequence
+from logging import Logger
+from typing import Any, overload
+from typing_extensions import TypeVarTuple, Unpack
+
+from transaction.interfaces import _DataManager, _Synchronizer, ISavepoint, ITransaction, ITransactionManager
+from transaction.weakset import WeakSet
+from zope.interface import implementer
+
+_Ts = TypeVarTuple("_Ts")
+
+class Status:
+    ACTIVE: str
+    COMMITTING: str
+    COMMITTED: str
+    DOOMED: str
+    COMMITFAILED: str
+
+class _NoSynchronizers:
+    @staticmethod
+    def map(_f: Callable[..., object]) -> None: ...
+
+@implementer(ITransaction)
+class Transaction:
+    status: str
+    extension: dict[str, Any]
+    log: Logger
+    def __init__(self, synchronizers: WeakSet[_Synchronizer] | None = None, manager: ITransactionManager | None = None) -> None: ...
+    @property
+    def user(self) -> str: ...
+    @user.setter
+    def user(self, v: str) -> None: ...
+    @property
+    def description(self) -> str: ...
+    @description.setter
+    def description(self, v: str) -> None: ...
+    def isDoomed(self) -> bool: ...
+    def doom(self) -> None: ...
+    def join(self, resource: _DataManager) -> None: ...
+    def savepoint(self, optimistic: bool = False) -> Savepoint: ...
+    def commit(self) -> None: ...
+    def getBeforeCommitHooks(self) -> Iterator[tuple[Callable[..., object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addBeforeCommitHook(self, hook: Callable[[], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addBeforeCommitHook(self, hook: Callable[[Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addBeforeCommitHook(self, hook: Callable[..., object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getAfterCommitHooks(self) -> Iterator[tuple[Callable[[bool, Unpack[tuple[Any, ...]]], object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addAfterCommitHook(self, hook: Callable[[bool], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addAfterCommitHook(self, hook: Callable[[bool, Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addAfterCommitHook(self, hook: Callable[[bool, Unpack[tuple[Any, ...]]], object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getBeforeAbortHooks(self) -> Iterator[tuple[Callable[..., object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addBeforeAbortHook(self, hook: Callable[[], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addBeforeAbortHook(self, hook: Callable[[Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addBeforeAbortHook(self, hook: Callable[..., object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getAfterAbortHooks(self) -> Iterator[tuple[Callable[..., object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addAfterAbortHook(self, hook: Callable[[], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addAfterAbortHook(self, hook: Callable[[Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addAfterAbortHook(self, hook: Callable[..., object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def data(self, ob: object) -> Any: ...
+    def set_data(self, ob: object, ob_data: object) -> None: ...
+    def abort(self) -> None: ...
+    def note(self, text: str | None) -> None: ...
+    def setUser(self, user_name: str, path: str = "/") -> None: ...
+    def setExtendedInfo(self, name: str, value: object) -> None: ...
+    def isRetryableError(self, error: BaseException) -> bool: ...
+
+def rm_key(rm: object) -> str | None: ...
+
+@implementer(ISavepoint)
+class Savepoint:
+    transaction: Transaction
+    def __init__(self, transaction: Transaction, optimistic: bool, *resources: _DataManager) -> None: ...
+    @property
+    def valid(self) -> bool: ...
+    def rollback(self) -> None: ...
+
+class AbortSavepoint:
+    datamanager: _DataManager
+    transaction: Transaction
+    def __init__(self, datamanager: _DataManager, transaction: Transaction) -> None: ...
+    def rollback(self) -> None: ...
+
+class NoRollbackSavepoint:
+    datamanager: _DataManager
+    def __init__(self, datamanager: _DataManager) -> None: ...
+    def rollback(self) -> None: ...
+
+def text_or_warn(s: object) -> str: ...

stubs/transaction/interfaces.pyi

@@ -0,0 +1,130 @@
+from collections.abc import Callable, Iterable, Sequence
+from typing import Any, Protocol, TypeVar, overload
+from typing_extensions import TypeVarTuple, Unpack
+
+from zope.interface import Interface
+
+_T = TypeVar("_T")
+_Ts = TypeVarTuple("_Ts")
+
+class _Attempt(Protocol):
+    def __enter__(self) -> ITransaction: ...
+    def __exit__(self, t: object, v: object, tb: object, /) -> bool | None: ...
+
+class _DataManager(Protocol):
+    @property
+    def transaction_manager(self) -> ITransactionManager: ...
+    def abort(self, transaction: ITransaction, /) -> object: ...
+    def tpc_begin(self, transaction: ITransaction, /) -> object: ...
+    def commit(self, transaction: ITransaction, /) -> object: ...
+    def tpc_vote(self, transaction: ITransaction, /) -> object: ...
+    def tpc_finish(self, transaction: ITransaction, /) -> object: ...
+    def tpc_abort(self, transaction: ITransaction, /) -> object: ...
+    def sortKey(self) -> str: ...
+
+class _Synchronizer(Protocol):
+    def beforeCompletion(self, transaction: ITransaction, /) -> object: ...
+    def afterCompletion(self, transaction: ITransaction, /) -> object: ...
+    def newTransaction(self, transaction: ITransaction, /) -> object: ...
+
+
+class ITransactionManager(Interface):
+    explicit: bool
+    def begin() -> ITransaction: ...
+    def get() -> ITransaction: ...
+    def commit() -> object: ...
+    def abort() -> object: ...
+    def doom() -> object: ...
+    def isDoomed() -> bool: ...
+    def savepoint(optimistic: bool = False) -> ISavepoint: ...
+    def registerSynch(synch: _Synchronizer) -> object: ...
+    def unregisterSynch(synch: _Synchronizer) -> object: ...
+    def clearSynchs() -> object: ...
+    def registeredSynchs() -> bool: ...
+    def attempts(number: int = 3) -> Iterable[_Attempt]: ...
+    @overload
+    def run(func: Callable[[], _T], tries: int = 3) -> _T: ...
+    @overload
+    def run(func: None = None, tries: int = 3) -> Callable[[Callable[[], _T]], _T]: ...
+    @overload
+    def run(tries: int, /) -> Callable[[Callable[[], _T]], _T]: ...
+
+class ITransaction(Interface):
+    user: str
+    description: str
+    extension: dict[str, Any]
+    def commit() -> object: ...
+    def abort() -> object: ...
+    def doom() -> object: ...
+    def savepoint(optimistic: bool = False) -> ISavepoint: ...
+    def join(datamanager: _DataManager) -> object: ...
+    def note(text: str) -> object: ...
+    def setExtendedInfo(name: str, value: object) -> object: ...
+    @overload
+    def addBeforeCommitHook(hook: Callable[[], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addBeforeCommitHook(hook: Callable[[Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addBeforeCommitHook(hook: Callable[..., object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getBeforeCommitHooks() -> Iterable[tuple[Callable[..., object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addAfterCommitHook(hook: Callable[[bool], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addAfterCommitHook(hook: Callable[[bool, Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addAfterCommitHook(hook: Callable[[bool, Unpack[tuple[Any, ...]]], object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getAfterCommitHooks() -> Iterable[tuple[Callable[[bool, Unpack[tuple[Any, ...]]], object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addBeforeAbortHook(hook: Callable[[], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addBeforeAbortHook(hook: Callable[[Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addBeforeAbortHook(hook: Callable[..., object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getBeforeAbortHooks() -> Iterable[tuple[Callable[..., object], Sequence[Any], dict[str, Any]]]: ...
+    @overload
+    def addAfterAbortHook(hook: Callable[[], object], args: tuple[()] = (), kws: None = None) -> object: ...
+    @overload
+    def addAfterAbortHook(hook: Callable[[Unpack[_Ts]], object], args: tuple[Unpack[_Ts]], kws: None = None) -> object: ...
+    @overload
+    def addAfterAbortHook(hook: Callable[..., object], args: Sequence[Any], kws: dict[str, Any]) -> object: ...
+    def getAfterAbortHooks() -> Iterable[tuple[Callable[..., object], Sequence[Any], dict[str, Any]]]: ...
+    def set_data(ob: object, data: object) -> object: ...
+    def data(ob: object) -> Any: ...
+    def isRetryableError(error: BaseException) -> bool: ...
+
+class IDataManager(Interface):
+    transaction_manager: ITransactionManager
+    def abort(transaction: ITransaction) -> object: ...
+    def tpc_begin(transaction: ITransaction) -> object: ...
+    def commit(transaction: ITransaction) -> object: ...
+    def tpc_vote(transaction: ITransaction) -> object: ...
+    def tpc_finish(transaction: ITransaction) -> object: ...
+    def tpc_abort(transaction: ITransaction) -> object: ...
+    def sortKey() -> str: ...
+
+class ISavepointDataManager(IDataManager):
+    def savepoint() -> IDataManagerSavepoint: ...
+
+class IRetryDataManager(IDataManager):
+    def should_retry(exception: BaseException) -> bool: ...
+
+class IDataManagerSavepoint(Interface):
+    def rollback() -> object: ...
+
+class ISavepoint(Interface):
+    def rollback() -> object: ...
+    valid: bool
+
+class InvalidSavepointRollbackError(Exception): ...
+
+class ISynchronizer(Interface):
+    def beforeCompletion(transaction: ITransaction) -> object: ...
+    def afterCompletion(transaction: ITransaction) -> object: ...
+    def newTransaction(transaction: ITransaction) -> object: ...
+
+class TransactionError(Exception): ...
+class TransactionFailedError(TransactionError): ...
+class DoomedTransaction(TransactionError): ...
+class TransientError(TransactionError): ...
+class NoTransaction(TransactionError): ...
+class AlreadyInTransaction(TransactionError): ...

stubs/transaction/weakset.pyi

@@ -0,0 +1,17 @@
+from collections.abc import Callable
+from typing import Generic, TypeVar
+
+from weakref import KeyedRef, WeakValueDictionary
+
+_T = TypeVar("_T")
+
+class WeakSet(Generic[_T]):
+    data: WeakValueDictionary[int, _T]
+    def __init__(self) -> None: ...
+    def __len__(self) -> int: ...
+    def __contains__(self, obj: object) -> bool: ...
+    def add(self, obj: _T) -> None: ...
+    def remove(self, obj: _T) -> None: ...
+    def clear(self) -> None: ...
+    def map(self, f: Callable[[_T], object]) -> None: ...
+    def as_weakref_list(self) -> list[KeyedRef[int, _T]]: ...

stubs/zope/sqlalchemy/__init__.pyi

@@ -0,0 +1,7 @@
+from zope.sqlalchemy.datamanager import (
+    ZopeTransactionEvents as ZopeTransactionEvents,
+    mark_changed as mark_changed,
+    register as register,
+)
+
+invalidate = mark_changed