sqlalchemy匹配Oracle数据库使用大写列名

公司的数据库是Oracle的，并且提供的数据字典是大小写混合表名，纯大写列名。

但是实际上，在公司数据库中，无论是表名还是列名都是大小写不敏感的，因此，sqlalchemy进行reflect的时候，
会将全部的表名和列名都转换为小写。这就导致从网页的数据字典复制到python代码中需要额外的将列名转换为小写，很不方便。

In Oracle, the data dictionary represents all case insensitive identifier names using UPPERCASE text. SQLAlchemy on the other hand considers an all-lower case identifier name to be case insensitive. The Oracle dialect converts all case insensitive identifiers to and from those two formats during schema level communication, such as reflection of tables and indexes. Using an UPPERCASE name on the SQLAlchemy side indicates a case sensitive identifier, and SQLAlchemy will quote the name - this will cause mismatches against data dictionary data received from Oracle, so unless identifier names have been truly created as case sensitive (i.e. using quoted names), all lowercase names should be used on the SQLAlchemy side.

solution 1

首先考虑将生成的Model的属性名改为大写，这样就不需要对sqlalchemy动刀，生成的model大概是这样：

class Mytablemodel(Base):
    __tablename__ = 'mytablemodel'
    __table_args__ = (
        PrimaryKeyConstraint('object_id', name='sys_c00200245'),
    )
    OBJECT_ID = Column('object_id', VARCHAR(100), comment='对象ID')
    ...

既能够在代码里列名和数据字典保持一致，又不需要大动干戈，何乐不为呢？

因此重写了sqlacodegen的generator:

class MyModelGenerator(DeclarativeGenerator):
    """将列名全部替换为大写"""
    def render_column_attribute(self, column_attr: ColumnAttribute) -> str:
        column_attr.name = column_attr.name.uppper()
        return super().render_column_attribute(column_attr)

但是，测试中发现了尴尬的情况：

stmt = select(Mytablemodel)
with engine.connect() as conn:
    with conn.begin():   # Optional: start a transaction
        q = conn.execute(stmt)
        print(q.keys())

哦豁，露馅了，keys跟着实际数据库表的列名走, 与model中定义的attribute无关，还是个小写。
要说不能用吧，也不至于，但是总觉得不够完美。解决的思路大概有两种，一种是将实际sqlalchemy reflect
出来的列名设置为大写，另一种就是将engine.execute返回的对象修改。

solution 2

重新回去分析sqlalchemy的源码，仔细研究后发现，这个大小写转换主要存在于以下两个函数：

def normalize_name(self, name):
    if name is None:
        return None

    name_lower = name.lower()
    name_upper = name.upper()

    if name_upper == name_lower:
        # name has no upper/lower conversion, e.g. non-european characters.
        # return unchanged
        return name
    elif name_upper == name and not (
        self.identifier_preparer._requires_quotes
    )(name_lower):
        # name is all uppercase and doesn't require quoting; normalize
        # to all lower case
        return name_lower
    elif name_lower == name:
        # name is all lower case, which if denormalized means we need to
        # force quoting on it
        return quoted_name(name, quote=True)
    else:
        # name is mixed case, means it will be quoted in SQL when used
        # later, no normalizes
        return name

def denormalize_name(self, name):
    if name is None:
        return None

    name_lower = name.lower()
    name_upper = name.upper()

    if name_upper == name_lower:
        # name has no upper/lower conversion, e.g. non-european characters.
        # return unchanged
        return name
    elif name_lower == name and not (
        self.identifier_preparer._requires_quotes
    )(name_lower):
        name = name_upper
    return name

那么，第一反应肯定是将这个步骤给屏蔽掉，而且Sqlalchemy贴心在OracleDialect中提供了这个选项:requires_name_normalize = True，改为False不就完事大吉了？ 但是实际操作之后发现, 对于Oracle
的dialect而言,这个设置无效. 只能重写函数.

当然,重写函数并不困难, 甚至不需要重写, 可以考虑用event的方法

from sqlalchemy import event

@event.listens_for(metadata, 'column_reflect')
def receive_column_reflect(inspector, table, column_info):
    "listen for the 'column_reflect' event"

    # ... (event handling logic) ...

得到的model变成了这样:

class Mytablemodel(Base):
    __tablename__ = 'mytablemodel'
    __table_args__ = (
        PrimaryKeyConstraint('OBJECT_ID', name='sys_c00200245'),
    )
    OBJECT_ID = Column(VARCHAR(100), comment='对象ID')

似乎就快成功了, 跑个测试之后发现, compile出来的SQL中, 列名都带了引号, 因为sqlalchemy默认纯小写是
大小写不敏感的, 纯大写需要被quote来表名这是个大小写敏感的列名,但是实际上在数据库中这是大小写不敏感的
因此SQL会出错.

重新分析compile的代码, 发现可以在dialect中进行修改:

class MyOracleIdentifierPreparer(OracleIdentifierPreparer):

    def _requires_quotes(self, value):
        """Return True if the given identifier requires quoting."""
        lc_value = value.lower()
        uc_value = value.upper()
        return (
            lc_value in self.reserved_words
            or value[0] in self.illegal_initial_characters
            or not self.legal_characters.match(util.text_type(value))
            or ((lc_value != value) and (uc_value != value)) # change here, allow all uppercaes
        )

class MyDialect(oracle.dialect):
    preparer = MyOracleIdentifierPreparer

目前运行良好.