def define_tables(cls, metadata):
     Table(
         quoted_name('t1', quote=True), metadata,
         Column('id', Integer, primary_key=True),
     )
     Table(
         quoted_name('t2', quote=True), metadata,
         Column('id', Integer, primary_key=True),
         Column('t1id', ForeignKey('t1.id'))
     )

    def test_named_alias_disable_quote(self):
        cte = select([literal(1).label("id")]).cte(
            name=quoted_name('CTE', quote=False))

        s1 = select([cte.c.id]).alias(
            name=quoted_name("DontQuote", quote=False))

        s = select([s1])
        self.assert_compile(
            s,
            'WITH CTE AS (SELECT :param_1 AS id) '
            'SELECT DontQuote.id FROM '
            '(SELECT CTE.id AS id FROM CTE) AS DontQuote'
        )

 def test_pickle_anon_label(self):
     q1 = _anonymous_label(quoted_name("x", True))
     for loads, dumps in picklers():
         q2 = loads(dumps(q1))
         assert isinstance(q2, _anonymous_label)
         eq_(str(q1), str(q2))
         eq_(q1.quote, q2.quote)

def view(name, metadata, selectable):
    if metadata.schema:
        full_name = metadata.schema + "." + name
    else:
        full_name = name
    t = table(quoted_name(name, None))
    t.metadata = metadata
    t.fullname = full_name
    t.schema = quoted_name(metadata.schema, None)

    for c in selectable.c:
        c._make_proxy(t)

    CreateView(t, selectable).execute_at('after-create', metadata)
    DropView(t).execute_at('before-drop', metadata)
    return t

    def test_reflect_lowercase_forced_tables(self):

        m2 = MetaData(testing.db)
        t2_ref = Table(quoted_name('t2', quote=True), m2, autoload=True)
        t1_ref = m2.tables['t1']
        assert t2_ref.c.t1id.references(t1_ref.c.id)

        m3 = MetaData(testing.db)
        m3.reflect(only=lambda name, m: name.lower() in ('t1', 't2'))
        assert m3.tables['t2'].c.t1id.references(m3.tables['t1'].c.id)

 def normalize_name(self, name):
     if name is None:
         return None
     if name.upper() == name and not \
             self.identifier_preparer._requires_quotes(name.lower()):
         return name.lower()
     elif name.lower() == name:
         return quoted_name(name, quote=True)
     else:
         return name

    def test_quoted_name_bindparam_ok(self):
        from sqlalchemy.sql.elements import quoted_name

        with testing.db.connect() as conn:
            eq_(
                conn.scalar(
                    select(
                        [cast(
                            literal(quoted_name("some_name", False)), String)]
                    )
                ),
                "some_name"
            )

 def normalize_name(self, name):
     # Remove trailing spaces: FB uses a CHAR() type,
     # that is padded with spaces
     name = name and name.rstrip()
     if name is None:
         return None
     elif name.upper() == name and \
             not self.identifier_preparer._requires_quotes(name.lower()):
         return name.lower()
     elif name.lower() == name:
         return quoted_name(name, quote=True)
     else:
         return name

    def check_db_version(self, db_version_info, session=None):
        """
        Checks the database version and prints an error message on database
        version mismatch.

        - On mismatching or on missing version a sys.exit(1) is called.
        - On missing DBVersion table, it returns False
        - On compatible DB version, it returns True

        Parameters:
            db_version_info (db_version.DBVersionInfo): required database
                version.
            session: an open database session or None. If session is None, a
                new session is created.
        """

        try:
            dispose_engine = False
            if session is None:
                engine = SQLServer.create_engine(self.get_connection_string())
                dispose_engine = True
                session = CreateSession(engine)
            else:
                engine = session.get_bind()

            if not engine.has_table(quoted_name(DBVersion.__tablename__,
                                                True)):
                LOG.debug("Missing DBVersion table!")
                return False

            version = session.query(DBVersion).first()
            if version is None:
                # Version is not populated yet
                LOG.error('No version information found in the database.')
                sys.exit(1)
            elif not db_version_info.is_compatible(version.major,
                                                   version.minor):
                LOG.error('Version mismatch. Expected database version: ' +
                          str(db_version_info))
                version_from_db = 'v' + str(version.major) + '.' + str(
                    version.minor)
                LOG.error('Version from the database is: ' + version_from_db)
                LOG.error('Please update your database.')
                sys.exit(1)

            LOG.debug("Database version is compatible.")
            return True
        finally:
            session.commit()
            if dispose_engine:
                engine.dispose()

    def normalize_name(self, name):
        """
        Converting EXASol case insensitive identifiers (upper case)
        to  SQLAlchemy case insensitive identifiers (lower case)
        """
        if name is None:
            return None

        if six.PY2:
            if isinstance(name, str):
                name = name.decode(self.encoding)

        if name.upper() == name and \
              not self.identifier_preparer._requires_quotes(name.lower()):
            return name.lower()
        elif name.lower() == name:
            return quoted_name(name, quote=True)
        else:
            return name

 def test_coerce_quoted_switch(self):
     q1 = quoted_name("x", False)
     q2 = quoted_name(q1, True)
     eq_(q2.quote, True)

 def test_rconcat_quotefalse(self):
     q1 = quoted_name("x", False)
     self._assert_not_quoted("y" + q1)

 def test_rconcat_quotetrue(self):
     q1 = quoted_name("x", True)
     self._assert_not_quoted("y" + q1)

 def test_apply_map_plain(self):
     q1 = _anonymous_label(quoted_name("x%s", None))
     q2 = q1.apply_map(("bar"))
     eq_(q2, "xbar")
     self._assert_not_quoted(q2)

 def test_pickle_quote(self):
     q1 = quoted_name("x", True)
     for loads, dumps in picklers():
         q2 = loads(dumps(q1))
         eq_(str(q1), str(q2))
         eq_(q1.quote, q2.quote)

 def test_apply_map_quoted(self):
     q1 = _anonymous_label(quoted_name("x%s", True))
     q2 = q1.apply_map(("bar"))
     eq_(q2, "xbar")
     eq_(q2.quote, True)

 def test_coerce_none(self):
     q1 = quoted_name(None, False)
     eq_(q1, None)

 def test_coerce_quoted_retain(self):
     q1 = quoted_name("x", False)
     q2 = quoted_name(q1, False)
     eq_(q2.quote, False)

 def test_rconcat_quotenone(self):
     q1 = quoted_name("x", None)
     self._assert_not_quoted("y" + q1)

 def test_rconcat_anon(self):
     q1 = _anonymous_label(quoted_name("x", True))
     assert isinstance(q1, _anonymous_label)
     value = "y" + q1
     assert isinstance(value, _anonymous_label)
     self._assert_quoted(value, True)