CODEWITHSUNDEEP
pythonsqlalchemyorm
Midnight
Midnight

Reputation: 423

How do I create relationships across different schemas in SQLAlchemy ORM?

I'm writing a Python application for an existing MySQL database ecosystem with tables spread across multiple schemas that I can't change without breaking a lot of legacy code (i.e. it's not an option). With my previous approach (see below) I was able to address all tables regardless of their schema so it didn't matter for the actual python code which schema a given table was in. However for my current project I want to implement some relationships that reach across different schemas and I noticed that this doesn't work. Is there a way to implement the different table classes in a way where the application using them doesn't need to know the schema and still support relationships between all of them?

from config import DB_USER, DB_PASS, DB_IP, DB_PORT
from sqlalchemy import create_engine, MetaData, Column, ForeignKey, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker, scoped_session, Session

DB_URI = f'mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_IP}:{DB_PORT}'
SchemaA = declarative_base()
SchemaB = declarative_base()
engineA = create_engine(DB_URI + '/schemaA')
engineB = create_engine(DB_URI + '/schemaB')


class RoutingSession(Session):
    def get_bind(self, mapper=None, clause=None):
        if mapper and issubclass(mapper.class_, SchemaA):
            return engineA
        elif mapper and issubclass(mapper.class_, SchemaB):
            return engineB


session_factory = sessionmaker(class_=RoutingSession)
Session = scoped_session(session_factory)
SchemaA.metadata = MetaData(bind=engineA)
SchemaB.metadata = MetaData(bind=engineB)


class Table1(SchemaA):
    __tablename__ = 'table1'
    __bind_key__ = 'schemaA'
    id = Column(Integer, primary_key=True)
    value = Column(String)


class Table2(SchemaB):
    __tablename__ = 'table2'
    __bind_key__ = 'schemaB'
    id = Column(Integer, primary_key=True)
    table1_id = Column(Integer, ForeignKey('schemaA.table1.id'))
    value = Column(String)
    # relationships
    tab1 = relationship('Table1', backref=backref('tab2'))


SchemaA.metadata.create_all()
SchemaB.metadate.create_all()

with Session() as session:
    t1 = session.query(Table1).get(1)  # This works
    t2 = t1.tab2                       # This doesn't work

I believe this has something to do with SchemaA and SchemaB not sharing the same declarative_base but I don't know how to fix this. Any ideas would be greatly appreciated. Also the example is a simplified version of my original project (that works) with changed names and the added relationship. I didn't run this code specifically, so it could potentially contain some errors and typos but the concept itself is sound.

edit: The error message I'm getting is:

sqlalchemy.exc.InvalidRequestError: When initializing mapper mapped class Table2->tab2, expression 'Table1' failed to locate a name ('Table1'). If this is a class name, consider adding this relationship() to the <class 'Table2'> class after both dependent classes have been defined.

I already tried the obvious solution of simply moving the class Table2 in front of the class Table1 but that didn't fix my problem (error message remained the same). Also in case it matters, in my non-simplified original code the engines, the RoutingSession and the Metadata binding all happen in a file database.py while the table classes are all defined in a file models.py importing Session and the schemas from database.

Upvotes: 2

Views: 3859

Answers (1)

Ian Wilson
Ian Wilson

Reputation: 9089

Are they on the same server? If so it might work.

Took me a long while to get mysql to startup in docker but I was able to test a lot of things, later including this: https://github.com/sqlalchemy/sqlalchemy/discussions/8027

Mentioned here: SQLAlchemy: JOIN between different databases AND using different files in a module

If you drop the engines, set the schema on each table and drop the schema from the db uri it seems to work. (DB_* vars are just for my local setup).


DB_USER="tester"
DB_PASS="secret"
DB_IP="testmysql"
DB_PORT="3306"

from sqlalchemy import (
    Integer,
    String,
    ForeignKey,
)
from sqlalchemy.schema import (
    Column,
)
from sqlalchemy.orm import backref, relationship, declarative_base, Session
from sqlalchemy import create_engine, MetaData, Column, ForeignKey, Integer, String

engine = create_engine(f'mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_IP}:{DB_PORT}')

Base = declarative_base()

class Table1(Base):
    __tablename__ = 'table1'
    __bind_key__ = 'schema_a'
    __table_args__ = {'schema': 'schema_a'}
    id = Column(Integer, primary_key=True)
    value = Column(String(15))


class Table2(Base):
    __tablename__ = 'table2'
    __bind_key__ = 'schema_b'
    __table_args__ = {'schema': 'schema_b'}
    id = Column(Integer, primary_key=True)
    table1_id = Column(Integer, ForeignKey('schema_a.table1.id'))
    value = Column(String(15))
    # relationships
    tab1 = relationship('Table1', backref=backref('tab2s'))

Base.metadata.create_all(engine)


with Session(engine) as session, session.begin():
    t1 = Table1(value='value number 1')
    t2 = Table2(value='another value', tab1=t1)
    session.add_all([t1, t2])

with Session(engine) as session, session.begin():
    t1 = session.query(Table1).first()
    assert t1.tab2s
    print (t1.tab2s)

OUTPUT

[<__main__.Table2 object at 0x7f6a1275cf10>]

Upvotes: 3

Related Questions