Колба SQLAlchemy динамически отражает объекты базы данных - PullRequest
Новая
       49

Колба SQLAlchemy динамически отражает объекты базы данных

0 голосов
/ 21 октября 2019

Я пишу новое приложение, которое подключается к старой базе данных. Пока я отражаю объекты базы данных, а не определяю их вручную в классах SQLA ORM. Я настроил свое приложение для фляги следующим образом: 

config = Config.get_config_for_env(env)
app = Flask(name)
app.config.from_object(config)
metadata = MetaData()
db = SQLAlchemy(metadata=metadata)
db.init_app(app)
app.db = db
app.app_context().push()

# Reflect DB
db.metadata.reflect(bind=db.engine, views=True)

Вызов выше отражает всю базу данных. Мои приложения обычно касаются нескольких таблиц одновременно, поэтому имеет смысл лениво отражать таблицы базы данных. Это можно сделать так: 

db.metadata.reflect(bind=db.engine, schema='MySchema', only=['MyTable'])

. Для этого мне нужно будет вставить слой, который гарантирует, что перед выполнением запроса схема и таблица будут отражены. Это добавляет сложности, так как все запросы должны пройти через другой слой кода. Есть ли способ неявного отображения схемы базы данных и таблицы по запросу при выполнении запроса?

Ответы [ 2 ]

0 голосов
/ 29 октября 2019

AFAIK, нет способа сделать это. Это можно сделать через тестовый класс. Примерно так: self.clone() клонирует объект: 

class TempDbApp(BaseApp):
    def __init__(self, env_src, name='TempDbApp', *args, **kwargs):
        super().__init__('t', name, *args, **kwargs)
        self.env_src = env_src
        self.logger = logging.getLogger(__name__)
        self.schemas = ['dbo']
        self.metadata = MetaData()

    def setup(self):
        super().setup()
        self.test_db_name = self.create_db()

    def teardown(self):
        # Do not drop db at end because pool
        super().teardown()
        self.metadata.drop_all(self.db.engine)
        for schema in self.schemas:
            if schema != 'dbo':
                self.db.engine.execute(DropSchema(schema))
        self.drop_db()

    def create_db(self):
        url = copy(self.db.engine.url)
        engine = create_engine(url, connect_args={'autocommit': True}, isolation_level='AUTOCOMMIT')
        res = engine.execute(f"exec dbo.usp_createShortLivedDB")
        tempdbname = res.fetchone()[0]
        res.close()
        engine.dispose()
        self.db.engine.url.database = tempdbname
        return tempdbname

    def drop_db(self):
        url = copy(self.db.engine.url)
        db = url.database
        url.database = 'master'
        engine = create_engine(url, connect_args={'autocommit': True}, isolation_level='AUTOCOMMIT')

        if database_exists(url):
            assert db != 'master'
            res = engine.execute(f"EXEC dbo.usp_deleteshortliveddb @dbname = '{db}'")
            res.close()

    def fetch_schemas(self):
        results = self.db.engine.execute('SELECT name FROM sys.schemas')
        for schema in results:
            self.schemas.append(schema[0])
        results.close()

    def create_schema(self, schema):
        with self.session() as sess:
            sess.execute(CreateSchema(schema))
        self.schemas.append(schema)

    @lru_cache(maxsize=2048)
    def clone(self, table, schema):
        if schema not in self.schemas:
            self.create_schema(schema)

        self.metadata.reflect(self.engine_src, schema=schema, only=[table])
        self.metadata.drop_all(self.db.engine)  # precautionary in case previous test didn't clean things up
        self.metadata.create_all(self.db.engine)

    @lru_cache(maxsize=2048)
    def get_table(self, table, schema, db=None):
        self.clone(table, schema)
        return super().get_table(table, schema, db)

    @lru_cache(maxsize=2048)
    def get_selectable(self, table, schema, db=None):
        self.clone(table, schema)
        return Table(table, self.db.metadata, schema=schema, autoload=True, autoload_with=self.db.get_engine(bind=db))

    @lazy
    def engine_src(self):
        conn_string = f'mssql+pymssql://user:pass@{self.env_src}-slo-sql-ds/mydb?charset=utf8'
        return create_engine(conn_string, isolation_level='AUTOCOMMIT')

    def start(self):
        raise Exception("You must not call this method - this app is for testing")

Затем можно создать тестовый класс с использованием множественного наследования: 

@final
class MyRealWorldClassTest(TempDbApp, MyRealWorldClass):
    pass
0 голосов
/ 21 октября 2019

Ну, если вы знаете имя нужной вам таблицы, вы можете сделать: 

table_to_work_with = Table("tablename", db.metadata, bind=db.engine, autoload=True)

И вы можете использовать sqlalchemy.inspect для получения имен таблиц, как описано здесь: Списоктаблицы базы данных с SQLAlchemy

...