How to provide 'on the fly' which postgresql schema i want to use on sqlalchemy queries?

[nickmetal]

I want to select schema name before i did a query.

In traditional sqlalchemy i do:

from sqlalchemy import create_engine
engine = create_engine(...).execution_options(schema_translate_map={None: 'some_schema_name'})

And all next queries for each models will be use such path:

select * from some_schema_name.table;

But i faced with error using aiopg:

from aiopg.sa import create_engine
engine = create_engine(...).execution_options 
AttributeError: 'Engine' object has no attribute 'execution_options'

I was trying to find this in docs, but not found.
Which way is correct to provide schema name in aiopg.sa?

aiopg==0.15.0
SQLAlchemy==1.2.16

Thank you!

[aio-libs-bot]

GitMate.io thinks possibly related issues are #45 (Possibly use SQLAlchemy Strategies), #140 (Using cursors with SQLAlchemy ), #345 (SQLAlchemy Dialects), #43 (Error with sqlalchemy), and #364 (close cannot be used while an asynchronous query is underway).

[vir-mir]

@nickmetal Hi!
we do not support execution_options

But you can set the scheme like this:

import sqlalchemy as sa

meta = sa.MetaData(schema='default_schema')
tbl = sa.Table('sa_tbl4', meta,
              sa.Column('id', sa.Integer, nullable=False, primary_key=True),
              sa.Column('name', sa.String(255), nullable=False,
                        default='test'))

def _str_sql_strip(sql):
   return str(sql).strip().replace('\n', '')

def test_default_schema():
   assert _str_sql_strip(tbl.select()).strip() == (
       'SELECT default_schema.sa_tbl4.id, default_schema.sa_tbl4.name '
       'FROM default_schema.sa_tbl4'
   )

if you want to change the scheme dynamically, I would do it like this

from contextlib import contextmanager

import sqlalchemy as sa

meta = sa.MetaData(schema='default_schema')
tbl = sa.Table('sa_tbl4', meta,
               sa.Column('id', sa.Integer, nullable=False, primary_key=True),
               sa.Column('name', sa.String(255), nullable=False,
                         default='test'))

def _str_sql_strip(sql):
    return str(sql).strip().replace('\n', '')

@contextmanager
def table_dynamic_schema(table, schema):
    default_schema = table.schema
    table.schema = schema
    yield table
    table.schema = default_schema

def test_dynamic_schema():
    with table_dynamic_schema(tbl, 'new_schema') as t:
        assert _str_sql_strip(t.select()).strip() == (
            'SELECT new_schema.sa_tbl4.id, new_schema.sa_tbl4.name '
            'FROM new_schema.sa_tbl4'
        )

    assert _str_sql_strip(tbl.select()).strip() == (
        'SELECT default_schema.sa_tbl4.id, default_schema.sa_tbl4.name '
        'FROM default_schema.sa_tbl4'
    )

[nickmetal]

@vir-mir thank you for your answer. I solved my problem with the quite similar way:)