How to force Sqlalchemy to point to a schema different from the default schema for a connection string

I have an oracle database connection with just a read access. When I connect to the database using collection string

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

This connect successful but the default database schema to this connect/SID point to PD_READONLY which has empty tables. The Schema with the right data/tables is PD_FULL_ACCESS.

Using any database visualizer, I can query the data in the PD_FULL_ACCESS using return the expected data.

SELECT date_created, date_modified from PD_FULL_ACCESS.department;

SELECT date_created, date_modified from department;

return an error [Code: 942, SQL State: 42000] ORA-00942: table or view does not exist [Script position: 102 - 105] which is valid because there is no such table in the default PD_READONLY.

My SQL schema is as follows:

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' "))}
    )

How can I force Sqlalchemy to point all queries e.g session.query(Department).all() to PD_FULL_ACCESS instead of the default PD_READONLY

I have tried specifying the schema/service_name in the connect string as:

DB_SERVICE_NAME="PD_READONLY"

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

This gives the error

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

I also tried:

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

which give the error

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

Both attempt above looks like there need to be a tnslister tweak on the database level. Any solution on the database level is not feasible in this case as the database is legacy application with many dependency and it cannot be changed/modified because of this case.

On the query level, I tried a naïve approach to add the schema to the query as:

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

got the error:

 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

The project is setup to use ORM rather than plan sql queries.

Solution

You can define a schema_translation_map when you create the engine. For example, if the model does not contain a specific schema

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

and we just do

engine = create_engine(connection_url)

then queries like

qry = select(Department)

will run against the default schema for the current user. However, if we do

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

then SQLAlchemy will build SQL statements that target the specified schema.