Python sqlalchemy 模块，event() 实例源码

def _fk_pragma_on_connect(dbapi_con, connection_record):
    ''' Code to execute every time the database connection is opened.

    Ref: http://docs.sqlalchemy.org/en/rel_0_9/core/event.html#sqlalchemy.event.listens_for
    '''
    # Support for foreign keys must be explicitly turned on
    # every time the database is opened.
    dbapi_con.execute('PRAGMA foreign_keys=ON')

    # Only uncomment these if you know what you are doing.
    # See the SQLite documentation for details.
    #dbapi_con.execute("PRAGMA journal_mode = MEMORY")
    #dbapi_con.execute("PRAGMA synchronous = OFF")
    #dbapi_con.execute("PRAGMA temp_store = MEMORY")
    #dbapi_con.execute("PRAGMA cache_size = 500000")

# This allows the file to be 'import'ed any number of times, but attempts to
# connect to the database only once.

def fix_sqlite():
    engine = db.engine

    @sqlalchemy.event.listens_for(engine, "connect")
    def _connect(dbapi_connection, connection_record):
        dbapi_connection.isolation_level = None

    @sqlalchemy.event.listens_for(engine, "begin")
    def _begin(conn):
        conn.execute("BEGIN")

def _include_sqlalchemy(obj):
    for module in sqlalchemy, sqlalchemy.orm:
        for key in module.__all__:
            if not hasattr(obj, key):
                setattr(obj, key, getattr(module, key))
    # Note: obj.Table does not attempt to be a SQLAlchemy Table class.
    obj.Table = _make_table(obj)
    obj.relationship = _wrap_with_default_query_class(obj.relationship)
    obj.relation = _wrap_with_default_query_class(obj.relation)
    obj.dynamic_loader = _wrap_with_default_query_class(obj.dynamic_loader)
    obj.event = event

def _include_sqlalchemy(obj):
    for module in sqlalchemy, sqlalchemy.orm:
        for key in module.__all__:
            if not hasattr(obj, key):
                setattr(obj, key, getattr(module, key))
    # Note: obj.Table does not attempt to be a SQLAlchemy Table class.
    obj.Table = _make_table(obj)
    obj.relationship = _wrap_with_default_query_class(obj.relationship)
    obj.relation = _wrap_with_default_query_class(obj.relation)
    obj.dynamic_loader = _wrap_with_default_query_class(obj.dynamic_loader)
    obj.event = event

def _include_sqlalchemy(obj):
    for module in sqlalchemy, sqlalchemy.orm:
        for key in module.__all__:
            if not hasattr(obj, key):
                setattr(obj, key, getattr(module, key))
    # Note: obj.Table does not attempt to be a SQLAlchemy Table class.
    obj.Table = _make_table(obj)
    obj.relationship = _wrap_with_default_query_class(obj.relationship)
    obj.relation = _wrap_with_default_query_class(obj.relation)
    obj.dynamic_loader = _wrap_with_default_query_class(obj.dynamic_loader)
    obj.event = event

def _include_sqlalchemy(obj):
    for module in sqlalchemy, sqlalchemy.orm:
        for key in module.__all__:
            if not hasattr(obj, key):
                setattr(obj, key, getattr(module, key))
    # Note: obj.Table does not attempt to be a SQLAlchemy Table class.
    obj.Table = _make_table(obj)
    obj.relationship = _wrap_with_default_query_class(obj.relationship)
    obj.relation = _wrap_with_default_query_class(obj.relation)
    obj.dynamic_loader = _wrap_with_default_query_class(obj.dynamic_loader)
    obj.event = event