sqlalchemy classical mapping relationship issue

Question

I am on a mission to finally learn sqlAlchemy so that I can reap the benefits in the years to come.

I am neck deep in the sqlalchemy documents and have been for the past two days. I am hell bent on learning the classical mapping way, instead of the declarative, bc the db I want to hook up to exists, and does not have a unique id column in all of it's tables. According to this article classical mapping is the way to go under such circumstances

I have been following the classical examples from the sqlalchemy site, but I just cannot seem to find the correct relationship configuration to get this to work.

Here is all my code:

engine = create_engine(
    "mssql+pyodbc://someaddress/test?driver=FreeTDS?TDS_version=8.0", echo=True)

metadata = MetaData(engine)

class User(object):

    def __repr__(self):
        return "<User(User_id='%s', name='%s', age='%s')>" % (
                            self.user_id, self.name, self.age)
class Email(object):

    def __repr__(self):
        return "<User(email_id='%s', address='%s', user_id='%s')>" % (
                            self.email_id, self.address, self.user_id)

users = Table('users', metadata,
    Column('user_id', Integer, primary_key=True),
    Column('name', String(40)),
    Column('age', Integer),
    schema='test.dbo.users')

mapper(User, users, properties={'Email': relationship(Email, primaryjoin=users.c.user_id==emails.c.user_id)})

emails = Table('emails', metadata,
    Column('email_id', Integer, primary_key=True),
    Column('address', String),
    Column('user_id', Integer, ForeignKey('test.dbo.users.user_id')),
    schema='test.dbo.emails')

mapper(Email, emails)

Session = sessionmaker(bind=engine)
session = Session()

mary = session.query(User, Email).filter(User.user_id == Email.user_id)

The pursuing error message makes it clear that it is the mapper / relationship that is the problem.

InvalidRequestError: One or more mappers failed to initialize - can't proceed with initialization of other mappers.  Original exception was: Could not determine join condition between parent/child tables on relationship User.Email - there are no foreign keys linking these tables.  Ensure that referencing columns are associated with a ForeignKey or ForeignKeyConstraint, or specify a 'primaryjoin' expression.

I have tried a long list of different things to try to remedy, but I just cannot get to the bottom of it.

Any pointers in the right direction would be much appreciated!

The sqlalchemy version I am on is;

'1.0.12'

Answer 1

I had better luck with:

Column('user_id', Integer, ForeignKey('users.user_id')),

I also reordered the table and mapping code slightly:

users = Table('users', metadata,
    Column('user_id', Integer, primary_key=True),
    Column('name', String(40)),
    Column('age', Integer))

emails = Table('emails', metadata,
    Column('email_id', Integer, primary_key=True),
    Column('address', String),
    # foreign key to table users
    Column('user_id', Integer, ForeignKey('users.user_id')))

mapper(Email, emails)

mapper(User, users, properties={
    'Email': relationship(
        Email, primaryjoin=users.c.user_id==emails.c.user_id)})

Answer 2

Since SQLAlchemy 1.4, classical mapping is called imperative mapping, and the syntax changes a little.

For clarity, I like to prefix my Table objects with table_, while when possible I prefix the table name with tb_ (which obviously is not possible here, but I left as is).

It is also possible to define the one-to-many relationship using the model class in the mapper.

Here is my take on the subject, following DDD principles and separation between the business domain and the persistence related stuff, such as id.

model.py

from dataclasses import dataclass

@dataclass
class User:
    # user_id is related with persistence layer
    # it will be made accessible with SQLAlchemy
    name: str
    age: int

@dataclass
class Email:
    user: User
    address: str

orm.py

from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy.orm import registry
from sqlalchemy.orm import relationship
from sqlalchemy.schema import MetaData
import model


metadata = MetaData()
mapper_registry = registry(metadata=metadata)

table_user = Table(
    "tb_user",
    mapper_registry.metadata,
    Column("user_id", Integer, primary_key=True, autoincrement=True),
    Column("name", String(40)),
    Column("age", Integer),
)

table_email = Table(
    "tb_email",
    mapper_registry.metadata,
    Column("email_id", Integer, primary_key=True, autoincrement=True),
    Column("user_id", ForeignKey("tb_user.user_id")),
    Column("address", String),
)

def start_mappers():
    mapper_registry.map_imperatively(model.Email, table_email)

    mapper_registry.map_imperatively(
        model.User,
        table_user,
        properties={
            "email": relationship(model.Email),
        },
    )