stubs/sqlalchemy/events.pyi

from . import event

class DDLEvents(event.Events):
    def before_create(self, target, connection, **kw): ...
    def after_create(self, target, connection, **kw): ...
    def before_drop(self, target, connection, **kw): ...
    def after_drop(self, target, connection, **kw): ...
    def before_parent_attach(self, target, parent): ...
    def after_parent_attach(self, target, parent): ...
    def column_reflect(self, inspector, table, column_info): ...

class PoolEvents(event.Events):
    def connect(self, dbapi_connection, connection_record): ...
    def first_connect(self, dbapi_connection, connection_record): ...
    def checkout(self, dbapi_connection, connection_record, connection_proxy): ...
    def checkin(self, dbapi_connection, connection_record): ...
    def reset(self, dbapi_connection, connection_record): ...
    def invalidate(self, dbapi_connection, connection_record, exception): ...
    def soft_invalidate(self, dbapi_connection, connection_record, exception): ...
    def close(self, dbapi_connection, connection_record): ...
    def detach(self, dbapi_connection, connection_record): ...
    def close_detached(self, dbapi_connection): ...

class ConnectionEvents(event.Events):
    def before_execute(self, conn, clauseelement, multiparams, params): ...
    def after_execute(self, conn, clauseelement, multiparams, params, result): ...
    def before_cursor_execute(
        self, conn, cursor, statement, parameters, context, executemany
    ): ...
    def after_cursor_execute(
        self, conn, cursor, statement, parameters, context, executemany
    ): ...
    def dbapi_error(self, conn, cursor, statement, parameters, context, exception): ...
    def handle_error(self, exception_context): ...
    def engine_connect(self, conn, branch): ...
    def set_connection_execution_options(self, conn, opts): ...
    def set_engine_execution_options(self, engine, opts): ...
    def engine_disposed(self, engine): ...
    def begin(self, conn): ...
    def rollback(self, conn): ...
    def commit(self, conn): ...
    def savepoint(self, conn, name): ...
    def rollback_savepoint(self, conn, name, context): ...
    def release_savepoint(self, conn, name, context): ...
    def begin_twophase(self, conn, xid): ...
    def prepare_twophase(self, conn, xid): ...
    def rollback_twophase(self, conn, xid, is_prepared): ...
    def commit_twophase(self, conn, xid, is_prepared): ...

class DialectEvents(event.Events):
    def do_connect(self, dialect, conn_rec, cargs, cparams): ...
    def do_executemany(self, cursor, statement, parameters, context): ...
    def do_execute_no_params(self, cursor, statement, context): ...
    def do_execute(self, cursor, statement, parameters, context): ...