Option to ignore extra keywords in an sqlalchemy Mapped Class constructor?

Question

Per below, I am trying initialize a sqlalchemy Mapped Class from a python dictionary that has extra keys. Is it possible to have the Mapped Class automatically ignore the extra keys instead of throwing an error? Likewise, can the Mapped Class have default values if the keys are not present?

from sqlalchemy import Column, Integer, String
class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String)

And here is the init part:

my_example_user = {'id'=1, 'name'='john', 'extra_key'= 1234}
User(**my_example_user)

Which throws an invalid key error

Thoughts?

Answer 1

The default constructor for ORM models looks like this (docstring removed for brevity):

def _declarative_constructor(self: Any, **kwargs: Any) -> None:
    cls_ = type(self)
    for k in kwargs:
        if not hasattr(cls_, k):
            raise TypeError(
                "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
            )
        setattr(self, k, kwargs[k])

so it's straightforward to define an __init__ method on Base to extend it, as shown in the other answers:

def __init__(self, **kwargs):
    super().__init__(**{k: v for k, v in kwargs.items() if hasattr(type(self), k)})

However we can dispense with the need for an explicit __init__ by replacing the default constructor directly. Given this function:

def lenient_constructor(self, **kwargs):
    cls_ = type(self)
    for k in kwargs:
        if not hasattr(cls_, k):
            print(f'Skipping invalid attr {k!r}')
            continue
        setattr(self, k, kwargs[k])

Then for pre-2.0 syntax we can set it as the default constructor for Base like this:

from sqlalchemy import orm

Base = orm.declarative_base(constructor=lenient_constructor)

or like this for 2.0+:

from sqlalchemy import orm

registry = orm.registry(constructor=lenient_constructor)


class Base(orm.DeclarativeBase):
    registry = registry

Answer 2

This is how I solved a similar issue, basically inspect the ORM model for the columns and filter the dict based on it.

from sqlalchemy.inspection import inspect
from sqlalchemy import Column, Integer, String

class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String)

VALID_COLUMNS = [column.key for column in inspect(User).columns]
my_example_user = {'id'=1, 'name'='john', 'extra_key'= 1234}
my_example_user = {k: v for k, v in my_example_user.items() if k in VALID_COLUMNS}

User(**my_example_user)

More info https://docs.sqlalchemy.org/en/20/core/inspection.html

Hope this helps.

Answer 3

If your model has relationships, you can use your model's Mapper object, as @eric-ihli mentioned. Here is another way (note the __init__ method):

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import backref, relationship

from my_app.db_models import Base


class Employee(Base):
    __tablename__ = "employee"

    id = Column(Integer, primary_key=True, autoincrement=True)
    department_id = Column(Integer, ForeignKey("department.id"), index=True)

    email = Column(String, unique=True, index=True, nullable=False)
    name = Column(String)

    department = relationship(
        "Department", backref=backref("employees", cascade="all, delete-orphan")
    )


    def __init__(self, **kwargs):
        allowed_args = self.__mapper__.class_manager  # returns a dict
        kwargs = {k: v for k, v in kwargs.items() if k in allowed_args}
        super().__init__(**kwargs)

This way, you can create an employee model like this:

from contextlib import closing
from my_app.db_models import Department, Employee, SessionLocal


with closing(SessionLocal()) as db:
    dept = db.query(Department).filter(Department.name == 'HR').first()
    employee = Employee(name='John Smith', email='john@smith.com', department=dept)
    db.add(employee)
    db.commit()

Answer 4

SQLAlchemy Mapper objects have an attrs property which is a dictionary of the names of the fields of your mapped class.

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import class_mapper
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    name = Column(String)

user = {
    'name': 'Eihli',
    'skill': 11
}

user_mapper = class_mapper(User)
mapped_user = User(**user)
# Boom! TypeError: 'skill' is an invalid keyword argument for User

mapped_user = User(**{
    k: v for k, v in user.items()
    if k in user_mapper.attrs.keys()
})
# Success!

No need to mess around with maintaining an exclude lists or mucking about with dict or getting in the way of super calls.

If you're trying to generate models with nested data, you'll have to do things a little different. Otherwise you'll get an "Unhashable type 'dict'" error.

Here's an example of a helper to inspect the mapper and get the keys of the relationships.

def from_json(model, data):
    mapper = class_mapper(model)
    keys = mapper.attrs.keys()
    relationships = inspect(mapper).relationships
    args = {k: v for k, v in data.items()
            if k in keys and k not in relationships}
    return model(**args)

Answer 5

Based on R Yakovlev's answer, you can make the list of the elements dynamic:

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    def __init__(self, **kwargs):
        keep_kwargs = {k: v for k, v in kwargs.items() if k in user_columns}
        super(User, self).__init__(**keep_kwargs)


user_columns = [_ for _ in User.__dict__.keys() if not _.startswith('_')]

I wanted to try find a way to embed the user_columns in the object, like with a @hybrid_property, yet not have it called every time it's used.

I expect that is possible but exceeded my time limit.

Answer 6

Also to pass extra keywords and call Base.__init__() method you can exclude extrakeys from super() and after that do what you want:

from sqlalchemy import Column, Integer, String

class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String)

     def __init__(self, **kwargs):
         extra_kw_list = ['key1', 'key2']
         super(User, self).__init__(**{x: y for x, y in kwargs.items()
                                       if x not in extra_kw_list})
         #do something you need here
         item1, item2 = kwargs['key1'], kwargs['key2']

Answer 7

Are we guaranteed that the __init__ of the superclass which is in place will never have other desired effects than setting the __dict__ entries? I didn't feel quite comfortable bypassing the superclass call completely, so my attempt at solving this was as follows, passing on only the entries which correspond to column names:

class User(Base):

    # ...

    def __init__(self, **entries):
        '''Override to avoid TypeError when passed spurious column names'''
        col_names = set([col.name for col in self.__table__.columns])
        superentries = {k : entries[k] for k in col_names.intersection(entries.keys())}
        super().__init__(**superentries)

Answer 8

In short, define constructor which does not pass arguments up to its superclass:

class User(Base):

    # ...

    def __init__(self, **entries):

        # NOTE: Do not call superclass
        #       (which is otherwise a default behaviour).
        #super(User, self).__init__(**entries)

        self.__dict__.update(entries)

I hit the same problem in transition from peewee which requires the opposite - to pass arguments to its superclass (and, therefore, constructor was already defined). So, I just tried commenting the line out and things start to work.

UPDATE

Also, make sure that entries do not contain (and, therefore, overwrite) any meta field in User class defined for SQLAlchemy defined, for example, those ORM relationships. It's kind of obvious (SQLAlchemy), but when mistake is made, it might not be easy to spot the problem.