如何强制 Sqlalchemy 指向与连接字符串的默认架构不同的架构

Question

我有一个只有读访问权限的 Oracle 数据库连接。当我使用集合字符串连接到数据库时

engine = create_engine(f"oracle+oracledb://{DB_USER}:{DB_PASSWORD}@{DB_URL{DB_PORT}/{DB_SID}") 

此连接成功，但此连接/SID 的默认数据库架构指向 PD_READONLY，其中包含空表。具有正确数据/表的架构是PD_FULL_ACCESS。

使用任何数据库可视化工具，我可以使用返回预期数据来查询PD_FULL_ACCESS中的数据。

SELECT date_created, date_modified from PD_FULL_ACCESS.department; 

SELECT date_created, date_modified from department;

返回错误

[Code: 942, SQL State: 42000]  ORA-00942: table or view does not exist [Script position: 102 - 105]

，这是有效的，因为默认的PD_READONLY中没有这样的表。

我的SQL架构如下：

class Department:
    __tablename__ = "department"
    __sa_dataclass_metadata_key__ = "sa"

    id: float = field(
        init=False, metadata={"sa": Column(NUMBER(15, 0, False), primary_key=True)}
    )
    label: str = field(metadata={"sa": Column(VARCHAR(255), nullable=False)})
    active: str = field(
        metadata={"sa": Column(CHAR(1), nullable=False, server_default=text("'Y' "))}
    )

我如何强制Sqlalchemy将所有查询例如

session.query(Department).all()

指向PD_FULL_ACCESS而不是默认的PD_READONLY

我尝试将连接字符串中的 schema/service_name 指定为：

DB_SERVICE_NAME =“PD_READONLY”

engine = create_engine(
    f"oracle+oracledb://{DB_USER}:{DB_PASSWORD}@{DB_URL}:{DB_PORT}/?service_name={DB_SERVICE_NAME}")

这给出了错误

sqlalchemy.exc.DatabaseError: (oracledb.exceptions.DatabaseError) DPY-4027: no configuration directory to search for tnsnames.ora

我也尝试过：

engine = create_engine(f"oracle+oracledb://{DB_USER}:{DB_PASSWORD}@{DB_URL}:{DB_PORT}/{DB_SERVICE_NAME}")

这给出了错误

SID "None" is not registered with the listener at host

上面的两次尝试看起来都需要在数据库级别进行 tnslister 调整。在这种情况下，任何数据库级别的解决方案都是不可行的，因为数据库是具有许多依赖性的遗留应用程序，因此无法更改/修改。

在查询级别，我尝试了一种简单的方法将架构添加到查询中：

db.query("PD_FULL_ACCESS.Department").all()

出现错误：

sqlalchemy.exc.ArgumentError: Textual column expression 'PD_FULL_ACCESS.Department' should be explicitly declared with text('PD_FULL_ACCESS.Department'), or use column('PD_FULL_ACCESS.Department') for more specificity

该项目设置为使用 ORM 而不是计划 sql 查询。

Answer 1

schema_translation_map

。例如，如果模型不包含特定模式

class Department(Base):
    __tablename__ = "department"
    id = Column(Integer, primary_key=True, autoincrement=False)
    name = Column(String(100), nullable=False)

我们就是这么做的

engine = create_engine(connection_url)

然后查询

qry = select(Department)

将针对当前用户的默认架构运行。但是，如果我们这样做

engine = create_engine(
    connection_url,
    execution_options={"schema_translate_map": {None: "PD_FULL_ACCESS"}},
)

然后 SQLAlchemy 将构建针对指定模式的 SQL 语句。

Answer 2

=

符号。你需要这样的东西：

f"oracle+oracledb://{DB_USER}:{DB_PASSWORD}@{DB_URL}:{DB_PORT}/?service_name={DB_SERVICE_NAME}"

第二个问题在错误消息本身中有解决方案。你尝试过吗？