Search code examples
pythonsqlalchemy

How to structure dataclass with cattrs on sqlalchemy classical mapping?


I am trying to structure back some classes using cattr in an application that I uses sqlalchemy. For some reason, after I commit a class to the db cannot structure back any nested dataclass. I get the error:

in emit_backref_from_collection_append_event
    child_state, child_dict = instance_state(child), instance_dict(child)
AttributeError: 'dict' object has no attribute '_sa_instance_state'

Here is a code example.

from dataclasses import dataclass
from sqlalchemy.orm import backref, relationship, registry, sessionmaker
from sqlalchemy import Table, Column, ForeignKey, create_engine
from sqlalchemy.sql.sqltypes import Integer, String
import cattr

from src.vlep.domain.model import Patient

mapper_registry = registry()

parents = Table(
    'parents', mapper_registry.metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('name', String(255)),
)

children = Table(
    'children', mapper_registry.metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('name', String(255)),
    Column('parent_id', ForeignKey('parents.id', ondelete='SET NULL'))
)


def start_mappers():
    children_mapper = mapper_registry.map_imperatively(
        Child,
        children,
    )
    parent_mapper = mapper_registry.map_imperatively(
        Parent,
        parents,
        properties={
            'children': relationship(
                children_mapper,
                backref=backref('parents'))
        }
    )


@dataclass
class Child:

    name: str


@dataclass
class Parent:

    name: str
    children: list


if __name__ == "__main__":

    engine = create_engine("sqlite:///:memory:")
    mapper_registry.metadata.create_all(engine)
    start_mappers()
    session = sessionmaker(bind=engine)()

    c1 = Child('Uguinho')
    c2 = Child('Luizinho')
    p = Parent('Fulano', [c1, c2])

    session.add(p)
    session.commit()

    try:
        # This works
        d = cattr.unstructure(p, Child)
        p2 = cattr.structure(d, Child)
        d2 = cattr.unstructure(p2, Child)
        print(d2)
    finally:
        # This does not
        d = cattr.unstructure(p, Parent)
        p2 = cattr.structure(d, Parent)
        d2 = cattr.unstructure(p2, Parent)
        print(d2)

By the error message I am guessing sqlalchemy is trying to check if the property 'children' on the parent_mapper is a list of Child and it is getting a dict instead (which it is in the unstructured state). I actually don't even know how sqlalchemy is doing this check. Again guessing, I imagine it will always do some checking for consistence. Anyway, I have now idea how to solve this.


Solution

  • So I did a little digging through the SQLAlchemy docs, and found & adapted an example that makes your code work:

    from dataclasses import dataclass
    from typing import List
    
    import cattr
    from sqlalchemy.orm import (
        relationship,
        sessionmaker,
        registry,
    )
    from sqlalchemy import Table, Column, ForeignKey, create_engine, MetaData
    from sqlalchemy.sql.sqltypes import Integer, String
    
    # commented this out for the example
    # from src.vlep.domain.model import Patient
    
    mapper_registry = registry()
    
    
    @dataclass
    class Child:
    
        name: str
    
    
    @dataclass
    class Parent:
    
        name: str
        children: List[Child]
    
    
    metadata_obj = MetaData()
    
    parent = Table(
        "parent",
        metadata_obj,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("name", String(255)),
    )
    
    child = Table(
        "child",
        metadata_obj,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("name", String(255)),
        Column("parent_id", ForeignKey("parent.id", ondelete="SET NULL")),
    )
    
    
    def start_mappers():
        mapper_registry.map_imperatively(
            Parent,
            parent,
            properties={"children": relationship(Child, backref="parent")},
        )
        mapper_registry.map_imperatively(Child, child)
    
    
    if __name__ == "__main__":
    
        engine = create_engine("sqlite:///:memory:")
        metadata_obj.create_all(engine)
        start_mappers()
        session = sessionmaker(bind=engine)()
    
        c1 = Child("Uguinho")
        c2 = Child("Luizinho")
        p = Parent("Fulano", [c1, c2])
    
        session.add(p)
        session.commit()
    
        try:
            # This works
            d = cattr.unstructure(p, Child)
            p2 = cattr.structure(d, Child)
            d2 = cattr.unstructure(p2, Child)
            print(d2)
        finally:
            # This now works
            d = cattr.unstructure(p, Parent)
            p2 = cattr.structure(d, Parent)
            d2 = cattr.unstructure(p2, Parent)
            print(d2)
    

    Output:

    {'name': 'Fulano'}
    {'name': 'Fulano', 'children': [{'name': 'Uguinho'}, {'name': 'Luizinho'}]}
    

    I think maybe one of the main issues was that you were declaring the "backref" as backref=backref('parents'), and the function wrapper there was unnecessary? Not really sure...

    In any case, the dataclasses are used now for the mapping properly (as far as I can tell). I hope this example works for you!