Source code for sqlalchemy_unchained.foreign_key

import inspect
import sqlalchemy as sa

from sqlalchemy.sql.type_api import TypeEngine as SQLAType
from typing import *

from .base_model import BaseModel as Model
from .model_registry import ModelRegistry
from .utils import snake_case

[docs]def foreign_key(*args, fk_col: Optional[str] = None, primary_key: bool = False, nullable: bool = False, ondelete: Optional[str] = None, onupdate: Optional[str] = None, **kwargs, ) -> sa.Column: """ Helper method to add a foreign key column to a model. For example:: class Post(db.Model): category_id = db.foreign_key('Category') category = db.relationship('Category', back_populates='posts') Is equivalent to:: class Post(db.Model): category_id = db.Column(db.Integer, db.ForeignKey(''), nullable=False) category = db.relationship('Category', back_populates='posts') Customizing all the things:: class Post(db.Model): _category_id = db.foreign_key('category_id', # db column name db.String, # db column type 'categories', # foreign table name fk_col='pk') # foreign key col name Would be equivalent to:: class Post(db.Model): _category_id = db.Column('category_id', db.String, db.ForeignKey('')) :param args: :func:`foreign_key` takes up to three positional arguments. Most commonly, you will only pass one argument, which should be the table name you're linking to (or indirectly, the model class/name works too). If you want to customize the column name the foreign key gets stored in the database under, then it must be the first string argument, and you must *also* supply the table name. You can customize the column type that gets used by passing it too, eg ``sa.Integer`` or ``sa.String(36)``. :param str fk_col: The column name of the primary key on the *opposite* side of the relationship (defaults to :attr:`sqlalchemy_unchained.ModelRegistry.default_primary_key_column`). :param bool primary_key: Whether or not this :class:`~sqlalchemy.Column` is a primary key. :param bool nullable: Whether or not this :class:`~sqlalchemy.Column` should be nullable. :param str ondelete: The cascade operation for `ON DELETE`. :param str onupdate: The cascade operation for `ON UPDATE`. :param kwargs: Any other kwargs to pass the :class:`~sqlalchemy.Column` constructor. """ return sa.Column(*_get_fk_col_args(args, fk_col, ondelete, onupdate), primary_key=primary_key, nullable=nullable, **kwargs)
def _get_fk_col_args( args: Union[List[Any], Tuple[Any, ...]], fk_col: Optional[str] = None, ondelete: Optional[str] = None, onupdate: Optional[str] = None, *, _default_col_type=sa.Integer, ): fk_col = fk_col or ModelRegistry().default_primary_key_column try: model_class = [x for x in args if inspect.isclass(x) and issubclass(x, Model)][0] except IndexError: model_class = None try: col_type = [x for x in args if isinstance(x, SQLAType) or (inspect.isclass(x) and issubclass(x, SQLAType))][0] except IndexError: col_type = _default_col_type str_args = [x for x in args if isinstance(x, str)] col_name = (str_args[0] if str_args and (len(str_args) == 2 or model_class) else None) table_name = str_args[0] if str_args else None if model_class: table_name = model_class.__tablename__ elif col_name: table_name = str_args[1] if not table_name: raise TypeError('Could not determine the table name to use. Please provide ' 'one as a positional argument.') if table_name != table_name.lower(): table_name = snake_case(table_name) args = [col_name] if col_name else [] args += [ col_type, sa.ForeignKey(table_name + '.' + fk_col, ondelete=ondelete, onupdate=onupdate), ] return args __all__ = [ 'foreign_key', ]