This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""Sync DAG specific permissions, if necessary"""
        from airflow.security.permissions import DAG_ACTIONS, resource_name_for_dag
        from airflow.www.fab_security.sqla.models import Action, Permission, Resource
        root_dag_id = dag.parent_dag.dag_id if dag.parent_dag else dag.dag_id
        def needs_perms(dag_id: str) -> bool:
            dag_resource_name = resource_name_for_dag(dag_id)
            for permission_name in DAG_ACTIONS:
                if not (
                    session.query(Permission)
                    .join(Action)
                    .join(Resource)
                    .filter(Action.name == permission_name)
                    .filter(Resource.name == dag_resource_name)
                    .one_or_none()
                ):
                    return True
            return False
        if dag.access_control or needs_perms(root_dag_id):
            self.log.debug("Syncing DAG permissions: %s to the DB", root_dag_id)
            from airflow.www.security import ApplessAirflowSecurityManager
            security_manager = ApplessAirflowSecurityManager(session=session)
            security_manager.sync_perm_for_dag(root_dag_id, dag.access_control)