Python sqlalchemy.util 模块，py3k() 实例源码

def _expect_warnings(exc_cls, messages, regex=True, assert_=True,
                     py2konly=False):

    if regex:
        filters = [re.compile(msg, re.I | re.S) for msg in messages]
    else:
        filters = messages

    seen = set(filters)

    real_warn = warnings.warn

    def our_warn(msg, exception, *arg, **kw):
        if not issubclass(exception, exc_cls):
            return real_warn(msg, exception, *arg, **kw)

        if not filters:
            return

        for filter_ in filters:
            if (regex and filter_.match(msg)) or \
                    (not regex and filter_ == msg):
                seen.discard(filter_)
                break
        else:
            real_warn(msg, exception, *arg, **kw)

    with mock.patch("warnings.warn", our_warn):
        yield

    if assert_ and (not py2konly or not compat.py3k):
        assert not seen, "Warnings were not seen: %s" % \
            ", ".join("%r" % (s.pattern if regex else s) for s in seen)

def _expect_warnings(exc_cls, messages, regex=True, assert_=True,
                     py2konly=False):

    if regex:
        filters = [re.compile(msg, re.I | re.S) for msg in messages]
    else:
        filters = messages

    seen = set(filters)

    real_warn = warnings.warn

    def our_warn(msg, exception, *arg, **kw):
        if not issubclass(exception, exc_cls):
            return real_warn(msg, exception, *arg, **kw)

        if not filters:
            return

        for filter_ in filters:
            if (regex and filter_.match(msg)) or \
                    (not regex and filter_ == msg):
                seen.discard(filter_)
                break
        else:
            real_warn(msg, exception, *arg, **kw)

    with mock.patch("warnings.warn", our_warn):
        yield

    if assert_ and (not py2konly or not compat.py3k):
        assert not seen, "Warnings were not seen: %s" % \
            ", ".join("%r" % (s.pattern if regex else s) for s in seen)

def _expect_warnings(exc_cls, messages, regex=True, assert_=True,
                     py2konly=False):

    if regex:
        filters = [re.compile(msg, re.I | re.S) for msg in messages]
    else:
        filters = messages

    seen = set(filters)

    real_warn = warnings.warn

    def our_warn(msg, exception, *arg, **kw):
        if not issubclass(exception, exc_cls):
            return real_warn(msg, exception, *arg, **kw)

        if not filters:
            return

        for filter_ in filters:
            if (regex and filter_.match(msg)) or \
                    (not regex and filter_ == msg):
                seen.discard(filter_)
                break
        else:
            real_warn(msg, exception, *arg, **kw)

    with mock.patch("warnings.warn", our_warn):
        yield

    if assert_ and (not py2konly or not compat.py3k):
        assert not seen, "Warnings were not seen: %s" % \
            ", ".join("%r" % (s.pattern if regex else s) for s in seen)

def _expect_warnings(exc_cls, messages, regex=True, assert_=True,
                     py2konly=False):

    if regex:
        filters = [re.compile(msg, re.I | re.S) for msg in messages]
    else:
        filters = messages

    seen = set(filters)

    real_warn = warnings.warn

    def our_warn(msg, exception, *arg, **kw):
        if not issubclass(exception, exc_cls):
            return real_warn(msg, exception, *arg, **kw)

        if not filters:
            return

        for filter_ in filters:
            if (regex and filter_.match(msg)) or \
                    (not regex and filter_ == msg):
                seen.discard(filter_)
                break
        else:
            real_warn(msg, exception, *arg, **kw)

    with mock.patch("warnings.warn", our_warn):
        yield

    if assert_ and (not py2konly or not compat.py3k):
        assert not seen, "Warnings were not seen: %s" % \
            ", ".join("%r" % (s.pattern if regex else s) for s in seen)

def assert_compile(self, clause, result, params=None,
                        checkparams=None, dialect=None,
                        checkpositional=None,
                        use_default_dialect=False,
                        allow_dialect_select=False):
        if use_default_dialect:
            dialect = default.DefaultDialect()
        elif dialect == None and not allow_dialect_select:
            dialect = getattr(self, '__dialect__', None)
            if dialect == 'default':
                dialect = default.DefaultDialect()
            elif dialect is None:
                dialect = config.db.dialect
            elif isinstance(dialect, basestring):
                dialect = create_engine("%s://" % dialect).dialect

        kw = {}
        if params is not None:
            kw['column_keys'] = params.keys()

        if isinstance(clause, orm.Query):
            context = clause._compile_context()
            context.statement.use_labels = True
            clause = context.statement

        c = clause.compile(dialect=dialect, **kw)

        param_str = repr(getattr(c, 'params', {}))

        if util.py3k:
            param_str = param_str.encode('utf-8').decode('ascii', 'ignore')
            print(("\nSQL String:\n" + util.text_type(c) + param_str).encode('utf-8'))
        else:
            print(("\nSQL String:\n" + util.text_type(c).encode('utf-8') + param_str))

        cc = re.sub(r'[\n\t]', '', util.text_type(c))

        eq_(cc, result, "%r != %r on dialect %r" % (cc, result, dialect))

        if checkparams is not None:
            eq_(c.construct_params(params), checkparams)
        if checkpositional is not None:
            p = c.construct_params(params)
            eq_(tuple([p[x] for x in c.positiontup]), checkpositional)