value = cls.serialize_value(
            value=value,
            key=key,
            task_id=task_id,
            dag_id=dag_id,
            run_id=run_id,
            map_index=map_index,
        )
        
        session.query(cls).filter(
            cls.key == key,
            cls.run_id == run_id,
            cls.task_id == task_id,
            cls.dag_id == dag_id,
            cls.map_index == map_index,
        ).delete()
        new = cast(Any, cls)(  
            dag_run_id=dag_run_id,
            key=key,
            value=value,
            run_id=run_id,
            task_id=task_id,
            dag_id=dag_id,
            map_index=map_index,
        )
        session.add(new)
        session.flush()
    @classmethod
    @provide_session
    def get_value(