динамическая схема sqlalchemy для объекта во время выполнения

Один из вариантов – отображать конкретные таблицы, зависящие от учетной записи. Вот документация SqlAlchemy по этому вопросу.

Альтернативно, вы можете создать таблицу со статическим атрибутом schema и обновить ее по мере необходимости во время выполнения и запустить запросы, которые вам нужны. Я не могу придумать не-грязный способ сделать это. Итак, вот грязный вариант

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

1. добавьте все таблицы, специфичные для учетной записи для списка.

2. • если таблицы выражены в декларативном синтаксисе, тогда вы должны изменить атрибут DeclarativeName.__table__.schema . Я не уверен, что вам нужно также изменить DeclarativeName.__table_args__['schema'] , но я думаю, это не повредит.
   • Если таблицы выражены в синтаксисе таблицы старого стиля, вам необходимо изменить атрибут Table.schema .

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

   • user_id = Column(ForeignKey('my_schema.user.id')) должен быть записан как user_id = Column(ForeignKey(User.id)) . Затем вы можете изменить схему User на my_new_schema . В противном случае во время запроса sqlalchemy будет запутан, потому что внешний ключ будет указывать на my_schema.user.id то время как запрос будет указывать на my_new_schema.user .

4. Я не уверен, что более сложные отношения могут быть выражены без использования простого текста, поэтому я предполагаю, что это ограничение для моего предлагаемого решения.

Вот пример, который я написал в терминале:

 >>> from sqlalchemy import Column, Table, Integer, String, select, ForeignKey >>> from sqlalchemy.orm import relationship, backref >>> from sqlalchemy.ext.declarative import declarative_base >>> B = declarative_base() >>> >>> class User(B): ... __tablename__ = 'user' ... __table_args__ = {'schema': 'first_schema'} ... id = Column(Integer, primary_key=True) ... name = Column(String) ... email = Column(String) ... >>> class Posts(B): ... __tablename__ = 'posts' ... __table_args__ = {'schema':'first_schema'} ... id = Column(Integer, primary_key=True) ... user_id = Column(ForeignKey(User.id)) ... text = Column(String) ... >>> str(select([User.id, Posts.text]).select_from(User.__table__.join(Posts))) 'SELECT first_schema."user".id, first_schema.posts.text \nFROM first_schema."user" JOIN first_schema.posts ON first_schema."user".id = first_schema.posts.user_id' >>> account_specific = [User, Posts] >>> for Tbl in account_specific: ... Tbl.__table__.schema = 'second_schema' ... >>> str(select([User.id, Posts.text]).select_from(User.__table__.join(Posts))) 'SELECT second_schema."user".id, second_schema.posts.text \nFROM second_schema."user" JOIN second_schema.posts ON second_schema."user".id = second_schema.posts.user_id' 

Как вы видите, тот же запрос ссылается на second_schema после обновления атрибута схемы таблицы.

Они устанавливаются статически. Для внешних ключей требуется одно и то же обращение, и у меня есть дополнительная проблема, поскольку у меня есть несколько схем, которые содержат несколько таблиц, поэтому я сделал это:

 from sqlalchemy.ext.declarative import declarative_base staging_dbase = declarative_base() model_dbase = declarative_base() def adjust_schemas(staging, model): for vv in staging_dbase.metadata.tables.values(): vv.schema = staging for vv in model_dbase.metadata.tables.values(): vv.schema = model def all_tables(): return staging_dbase.metadata.tables.union(model_dbase.metadata.tables) 

Затем в моем стартовом коде:

 adjust_schemas(staging=staging_name, model=model_name) 

Вы можете изменить это для одной декларативной базы.