Relations

How do define relationships without explicitly setting foreign keys.

import sqlalchemy as db

from sqlalchemy import create_engine, text
from sqlalchemy.orm import declarative_base, relationship, Session
Base = declarative_base()

class User(Base):
    __tablename__ = "user"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    products = relationship("Product", back_populates="user")

class Product(Base):
    __tablename__ = "product"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
    user = relationship("User", back_populates="products")

db_path = "sqlite:///test.db"
engine = db.create_engine(db_path)
Base.metadata.create_all(engine, checkfirst=True)

session = Session(bind=engine)
# enforce foreign key constraint in SQLite (off by default!)
session.execute(text("PRAGMA foreign_keys=on"));

user = User(name="Alice")
product1 = Product(name="A", user=user)
product2 = Product(name="B", user=user)

session.add(user)
session.commit()

Foreign key will be correctly assigned by sqlalchemy:

product1.user_id

1

Works also the other way round:

user = User(name="Kyle")
product3 = Product(name="A")
product4 = Product(name="B")

user.products = [product3, product4]

session.add(user)
session.commit()

Foreign key will be correctly assigned by sqlalchemy:

product3.user_id

2

This does not work

The associated user.id will not be set correctly. It will be set to None which is the current value of user.id prior to committing.

user = User(name="Dave")

product5 = Product(name="C", user_id=user.id) # user.id is None at this point!
product6 = Product(name="D", user_id=user.id)

session.add_all([user, product5, product6])
session.commit()

---------------------------------------------------------------------------
IntegrityError                            Traceback (most recent call last)
File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:2115, in Connection._exec_insertmany_context(self, dialect, context)
   2114     else:
-> 2115         dialect.do_execute(
   2116             cursor,
   2117             sub_stmt,
   2118             sub_params,
   2119             context,
   2120         )
   2122 except BaseException as e:

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/default.py:942, in DefaultDialect.do_execute(self, cursor, statement, parameters, context)
    941 def do_execute(self, cursor, statement, parameters, context=None):
--> 942     cursor.execute(statement, parameters)

IntegrityError: NOT NULL constraint failed: product.user_id

The above exception was the direct cause of the following exception:

IntegrityError                            Traceback (most recent call last)
Cell In[8], line 7
      4 product6 = Product(name="D", user_id=user.id)
      6 session.add_all([user, product5, product6])
----> 7 session.commit()

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/session.py:2032, in Session.commit(self)
   2029 if trans is None:
   2030     trans = self._autobegin_t()
-> 2032 trans.commit(_to_root=True)

File <string>:2, in commit(self, _to_root)

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/state_changes.py:139, in _StateChange.declare_states.<locals>._go(fn, self, *arg, **kw)
    137 self._next_state = _StateChangeStates.CHANGE_IN_PROGRESS
    138 try:
--> 139     ret_value = fn(self, *arg, **kw)
    140 except:
    141     raise

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/session.py:1313, in SessionTransaction.commit(self, _to_root)
   1311 if self._state is not SessionTransactionState.PREPARED:
   1312     with self._expect_state(SessionTransactionState.PREPARED):
-> 1313         self._prepare_impl()
   1315 if self._parent is None or self.nested:
   1316     for conn, trans, should_commit, autoclose in set(
   1317         self._connections.values()
   1318     ):

File <string>:2, in _prepare_impl(self)

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/state_changes.py:139, in _StateChange.declare_states.<locals>._go(fn, self, *arg, **kw)
    137 self._next_state = _StateChangeStates.CHANGE_IN_PROGRESS
    138 try:
--> 139     ret_value = fn(self, *arg, **kw)
    140 except:
    141     raise

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/session.py:1288, in SessionTransaction._prepare_impl(self)
   1286     if self.session._is_clean():
   1287         break
-> 1288     self.session.flush()
   1289 else:
   1290     raise exc.FlushError(
   1291         "Over 100 subsequent flushes have occurred within "
   1292         "session.commit() - is an after_flush() hook "
   1293         "creating new objects?"
   1294     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/session.py:4353, in Session.flush(self, objects)
   4351 try:
   4352     self._flushing = True
-> 4353     self._flush(objects)
   4354 finally:
   4355     self._flushing = False

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/session.py:4488, in Session._flush(self, objects)
   4485     transaction.commit()
   4487 except:
-> 4488     with util.safe_reraise():
   4489         transaction.rollback(_capture_exception=True)

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/util/langhelpers.py:146, in safe_reraise.__exit__(self, type_, value, traceback)
    144     assert exc_value is not None
    145     self._exc_info = None  # remove potential circular references
--> 146     raise exc_value.with_traceback(exc_tb)
    147 else:
    148     self._exc_info = None  # remove potential circular references

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/session.py:4449, in Session._flush(self, objects)
   4447 self._warn_on_events = True
   4448 try:
-> 4449     flush_context.execute()
   4450 finally:
   4451     self._warn_on_events = False

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/unitofwork.py:466, in UOWTransaction.execute(self)
    464 else:
    465     for rec in topological.sort(self.dependencies, postsort_actions):
--> 466         rec.execute(self)

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/unitofwork.py:642, in SaveUpdateAll.execute(self, uow)
    640 @util.preload_module("sqlalchemy.orm.persistence")
    641 def execute(self, uow):
--> 642     util.preloaded.orm_persistence.save_obj(
    643         self.mapper,
    644         uow.states_for_mapper_hierarchy(self.mapper, False, False),
    645         uow,
    646     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/persistence.py:93, in save_obj(base_mapper, states, uowtransaction, single)
     81     update = _collect_update_commands(
     82         uowtransaction, table, states_to_update
     83     )
     85     _emit_update_statements(
     86         base_mapper,
     87         uowtransaction,
   (...)
     90         update,
     91     )
---> 93     _emit_insert_statements(
     94         base_mapper,
     95         uowtransaction,
     96         mapper,
     97         table,
     98         insert,
     99     )
    101 _finalize_insert_update_commands(
    102     base_mapper,
    103     uowtransaction,
   (...)
    119     ),
    120 )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/orm/persistence.py:1143, in _emit_insert_statements(base_mapper, uowtransaction, mapper, table, insert, bookkeeping, use_orm_insert_stmt, execution_options)
   1140 if do_executemany:
   1141     multiparams = [rec[2] for rec in records]
-> 1143     result = connection.execute(
   1144         statement, multiparams, execution_options=execution_options
   1145     )
   1147     if use_orm_insert_stmt is not None:
   1148         if return_result is None:

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:1416, in Connection.execute(self, statement, parameters, execution_options)
   1414     raise exc.ObjectNotExecutableError(statement) from err
   1415 else:
-> 1416     return meth(
   1417         self,
   1418         distilled_parameters,
   1419         execution_options or NO_OPTIONS,
   1420     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/sql/elements.py:523, in ClauseElement._execute_on_connection(self, connection, distilled_params, execution_options)
    521     if TYPE_CHECKING:
    522         assert isinstance(self, Executable)
--> 523     return connection._execute_clauseelement(
    524         self, distilled_params, execution_options
    525     )
    526 else:
    527     raise exc.ObjectNotExecutableError(self)

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:1638, in Connection._execute_clauseelement(self, elem, distilled_parameters, execution_options)
   1626 compiled_cache: Optional[CompiledCacheType] = execution_options.get(
   1627     "compiled_cache", self.engine._compiled_cache
   1628 )
   1630 compiled_sql, extracted_params, cache_hit = elem._compile_w_cache(
   1631     dialect=dialect,
   1632     compiled_cache=compiled_cache,
   (...)
   1636     linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
   1637 )
-> 1638 ret = self._execute_context(
   1639     dialect,
   1640     dialect.execution_ctx_cls._init_compiled,
   1641     compiled_sql,
   1642     distilled_parameters,
   1643     execution_options,
   1644     compiled_sql,
   1645     distilled_parameters,
   1646     elem,
   1647     extracted_params,
   1648     cache_hit=cache_hit,
   1649 )
   1650 if has_events:
   1651     self.dispatch.after_execute(
   1652         self,
   1653         elem,
   (...)
   1657         ret,
   1658     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:1841, in Connection._execute_context(self, dialect, constructor, statement, parameters, execution_options, *args, **kw)
   1838 context.pre_exec()
   1840 if context.execute_style is ExecuteStyle.INSERTMANYVALUES:
-> 1841     return self._exec_insertmany_context(dialect, context)
   1842 else:
   1843     return self._exec_single_context(
   1844         dialect, context, statement, parameters
   1845     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:2123, in Connection._exec_insertmany_context(self, dialect, context)
   2115         dialect.do_execute(
   2116             cursor,
   2117             sub_stmt,
   2118             sub_params,
   2119             context,
   2120         )
   2122 except BaseException as e:
-> 2123     self._handle_dbapi_exception(
   2124         e,
   2125         sql_util._long_statement(sub_stmt),
   2126         sub_params,
   2127         cursor,
   2128         context,
   2129         is_sub_exec=True,
   2130     )
   2132 if engine_events:
   2133     self.dispatch.after_cursor_execute(
   2134         self,
   2135         cursor,
   (...)
   2139         context.executemany,
   2140     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:2352, in Connection._handle_dbapi_exception(self, e, statement, parameters, cursor, context, is_sub_exec)
   2350 elif should_wrap:
   2351     assert sqlalchemy_exception is not None
-> 2352     raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
   2353 else:
   2354     assert exc_info[1] is not None

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/base.py:2115, in Connection._exec_insertmany_context(self, dialect, context)
   2113             break
   2114     else:
-> 2115         dialect.do_execute(
   2116             cursor,
   2117             sub_stmt,
   2118             sub_params,
   2119             context,
   2120         )
   2122 except BaseException as e:
   2123     self._handle_dbapi_exception(
   2124         e,
   2125         sql_util._long_statement(sub_stmt),
   (...)
   2129         is_sub_exec=True,
   2130     )

File ~/miniconda3/envs/satpy/lib/python3.13/site-packages/sqlalchemy/engine/default.py:942, in DefaultDialect.do_execute(self, cursor, statement, parameters, context)
    941 def do_execute(self, cursor, statement, parameters, context=None):
--> 942     cursor.execute(statement, parameters)

IntegrityError: (sqlite3.IntegrityError) NOT NULL constraint failed: product.user_id
[SQL: INSERT INTO product (name, user_id) VALUES (?, ?) RETURNING id]
[parameters: ('C', None)]
(Background on this error at: https://sqlalche.me/e/20/gkpj)