using sqlalchemy for airflow tasks , problem with sqlalchemy relationship

Viewed 24

i have an airflow file structure like below

dags/
|-- dag_file.py
|
plugins/
|-- my_task_1/
|   |-- tasks/
|   |   |-- task_file.py
|   |-- models/
|   |   |-- a.py
|   |   |-- b.py
|   |-- db/
|   |   |-- session.py
|   |   |-- base_class.py
|   |   |-- base.py

and in my base_class i simply create a declaritve base model for my models

metadata = MetaData()

@as_declarative(metadata=metadata)
class BaseModel(object):
    __abstract__ = True
    __table_args__ = {'extend_existing': True}
    id = Column(Integer, primary_key=True, autoincrement=True)

and in base i will import all models and base_model for session to have them all

from my_task_1.db.base_class import BaseModel
from my_task_1.db.models.a import A
from my_task_1.db.models.b import B

and in session i'll try to create a session from an airflow hook connection

from my_task_1.db import base

def get_session() -> Session:
    hook = OdbcHook(odbc_conn_id="test_db")
    engine = hook.get_sqlalchemy_engine()
    session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    return session()

everything works fine and i can create tables wirh metadata.create_all() and query models and tables as expected in my task instances but as soon as i try to add a relationship from any model to others like

#models/b.py

class B(BaseModel):
    ...
    foo = relationship("my_task_1.models.a.A", backref="a_lookup")

in any task that any of the two related models have been imported , sqlalchemy raises error

sqlalchemy.exc.ArgumentError: Error creating backref 'a_lookup' on relationship 'B.foo': property of that name exists on mapper 'mapped class A->a'

and i can confirm that there is not any property with those names in my other class

how can i fix this problem?

0 Answers
Related