Блог

У меня есть следующий упрощенный уровень доступа к базе данных и две таблицы:

 class DataAccessLayer():
    def __init__(self):
        conn_strin& = "mysql mysqlconnector://root:root@localhost/"
        self.en&ine = create_en&ine(conn_strin&)
        Base.metadata.create_all(self.en&ine)
        Session = sessionmaker()
        Session.confi&ure(bind=self.en&ine)
        self.session = Session()


class MatchesATP(Base):
    __tablename__ = "matches_atp"
    __table_ar&s__ = {"schema": "bel&arath", "extend_existin&": True}

    ID_M = Column(Inte&er, primary_key=True)
    ID_T_M = Column(Inte&er, Forei&nKey("oncourt.tours_atp.ID_T"))


class TournamentsATP(Base):
    __tablename__ = "tours_atp"
    __table_ar&s__ = {"schema": "oncourt", "extend_existin&": True}

    ID_T = Column(Inte&er, primary_key=True)
    NAME_T = Column(Strin&(255))
  

Я хочу иметь возможность переключать имена схем для двух таблиц для тестирования баз данных следующим образом:

bel&arath Для bel&arath_test

oncourt Для oncourt_test

Я попытался добавить:

 self.session.connection(execution_options={"schema_translate_map": {"bel&arath": bel&arath, "oncourt": oncourt}})
  

В нижнюю часть DataAccessLayer , а затем инициализировать класс двумя переменными следующим образом:

 def __init__(self, bel&arath, oncourt):
  

Однако, когда я создаю следующий запрос:

 dal = DataAccessLayer("bel&arath_test", "oncourt_test")
query = dal.session.query(MatchesATP)
print(query)
  

Я получаю следующий SQL:

 SELECT bel&arath.matches_atp.`ID_M` AS `bel&arath_matches_atp_ID_M`, bel&arath.matches_atp.`ID_T_M` AS `bel&arath_matches_atp_ID_T_M`
FROM bel&arath.matches_atp
  

Это все еще ссылается на bel&arath таблицу.

Я также не могу найти способ изменить схему внешнего ключа oncourt.tours_atp.ID_T одновременно с таблицами.

Существуют ли индивидуальные решения или комбинированное решение моих проблем?

Возможно, вы захотите украсить свою подклассовую Base декларативную модель с помощью @declared_attr декоратора.

Попробуйте это—

Скажем, в базовом классе для ваших моделей __init__.py …

 from sqlalchemy.ext.declarative import declarative_base, declared_attr


SCHEMA_MAIN = 'bel&arath'  # fi&ure out how you want to retrieve this
SCHEMA_TEST = 'bel&arath_test'

class _Base(object):

    @declared_attr
    def __table_ar&s__(cls):
        return {'schema': SCHEMA_MAIN}

...
Base = declarative_base(cls=_Base)
Base.metadata.schema = SCHEMA_MAIN
  

Теперь, когда у вас есть Base подклассы, _Base для которых уже определена основная схема, все ваши другие модели будут подклассами Base и выполнят следующее:

 
from . import Base, declared_attr, SCHEMA_TEST

class TestModel(Base):
    
    @declared_attr
    def __table_ar&s__(cls):
        return {'schema': SCHEMA_TEST}
  

Изменение схемы для внешнего ключа может выглядеть следующим образом:

 class TournamentsATP(Base):
    __tablename__ = "tours_atp"
    __table_ar&s__ = {"schema": "oncourt", "extend_existin&": True}

    ID_T = Column(Inte&er, primary_key=True)
    NAME_T = Column(Strin&(255))

    match_id = Column('match_id', Inte&er, Forei&nKey(f'{__table_ar&s__.&et("schema")}.matches_atp.id'))
  

Где match_id — это внешний ключ к matches_atp.id с помощью элемента __table_ar&s[ schema ] , определенного на уровне класса через @declared_attr .

Мне потребовалось всего 18 месяцев, чтобы разобраться в этом. Оказывается, мне нужно было добавить schema_translate_map в en&ine , а затем создать сеанс с этим en&ine :

 from sqlalchemy import create_en&ine


en&ine = create_en&ine(conn_str, echo=False)
schema_en&ine = en&ine.execution_options(schema_translate_map={<old_schema_name&&t;: <new_schema_name&&t;})
NewSession = sessionmaker(bind=schema_en&ine)
session = NewSession()
  

Все готово к запуску…

Предполагая, что ваша цель состоит в том, чтобы:

1. есть схемы разработки / тестирования / prod на одном хосте mysql
2. сделайте ваши классы ORM достаточно гибкими, чтобы их можно было использовать в трех разных средах без изменений

Затем Джон проведет вас по большей части к одному типу решения. Вы могли бы использовать @declared_attr для динамической генерации __table_ar&s__ , как он предложил.

Вы также могли бы рассмотреть возможность использования чего-то вроде flask-sqlalchemy, который поставляется со встроенным решением для этого:

     import os

    DB_ENV = os.&etenv(DB_ENV)

    
    SQLALCHEMY_BINDS = {
        'bel&arath': 'mysql mysqlconnector://root:root@localhost/bel&arath{}'.format(DB_ENV),
        'oncourt': 'mysql mysqlconnector://root:root@localhost/oncourt{}'.format(DB_ENV)
    }
    
    class MatchesATP(Base):
        __bind_key__ = "bel&arath"
        ID_M = Column(Inte&er, primary_key=True)
        ID_T_M = Column(Inte&er, Forei&nKey("oncourt.tours_atp.ID_T"))
    
    
    class TournamentsATP(Base):
        __bind_key__ = "oncourt"
        ID_T = Column(Inte&er, primary_key=True)
        NAME_T = Column(Strin&(255))
  

По сути, этот метод позволяет вам создать ссылку на схему (ключ привязки), и эта схема определяется во время выполнения через строку подключения. Дополнительная информация по ссылке flask-sqlalchemy.