Did I find the right examples for you? yes no      Crawl my project      Python Jobs

All Samples(7)  |  Call(5)  |  Derive(0)  |  Import(2)

src/c/a/cache-busters-HEAD/cache_buster/tap.py   cache-busters(Download)
 
from cache_buster.caches import MemcachedCache, MultiCache
from cache_buster.databases import (
    DatabaseListenerService, MySQLDatabaseListener
)
    connection = BinLogStreamReader(connection_settings=config['database'])
 
    listener = MySQLDatabaseListener(reactor, connection, driver, log)
 
    return DatabaseListenerService(cooperator, listener, log)

src/c/a/cache-busters-HEAD/cache_buster/test/test_databases.py   cache-busters(Download)
from twisted.trial.unittest import TestCase
 
from cache_buster.databases import MySQLDatabaseListener
from cache_buster.test.doubles import DummyLogger
 
            invalidate_row=pretend.call_recorder(lambda table, row: d)
        )
        listener = MySQLDatabaseListener(None, connection, driver, DummyLogger())
        iterable = iter(listener)
        res = next(iterable)
            invalidate_row=pretend.call_recorder(lambda table, row: d)
        )
        listener = MySQLDatabaseListener(None, connection, driver, DummyLogger())
        iterable = iter(listener)
        res = next(iterable)
    def test_ignores_unknown_event_types(self):
        connection = pretend.stub(
            fetchone=lambda: self.create_event(None, None, None)
        )
        listener = MySQLDatabaseListener(None, connection, None, DummyLogger())
            invalidate_row=pretend.call_recorder(lambda table, row: None)
        )
        listener = MySQLDatabaseListener(None, connection, driver, DummyLogger())
        iterable = iter(listener)
        next(iterable)