'Process a Python file containing Airflow DAGs.
This includes:
1. Execute the file and look for DAG objects in the namespace.
2. Pickle the DAG and save it to the DB (if necessary).
3. For each DAG, see what tasks should run and create appropriate task
instances in the DB.
4. Record any errors importing the file into ORM
5. Kill (in ORM) any task instances belonging to the DAGs that haven\'t
issued a heartbeat in a while.
Returns a list of SimpleDag objects that represent the DAGs found in
the file
:param file_path: the path to the Python file that should be executed
:type file_path: unicode
:param pickle_dags: whether serialize the DAGs found in the file and
save them to the db
:type pickle_dags: bool
:return: a list of SimpleDags made from the Dags found in the file
:rtype: list[SimpleDag]'

@provide_session
def process_file(self, file_path, pickle_dags=False, session=None):

    self.logger.info(u'Processing    file    {}    for    tasks    to    queue'.format(file_path))
    simple_dags = []
    try:
        dagbag = models.DagBag(file_path)
    except Exception:
        self.logger.exception(u'Failed    at    reloading    the    DAG    file    {}'.format(file_path))
        Stats.incr(u'dag_file_refresh_error', 1, 1)
        return []
    if (len(dagbag.dags) > 0):
        self.logger.info(u'DAG(s)    {}    retrieved    from    {}'.format(dagbag.dags.keys(), file_path))
    else:
        self.logger.warning(u'No    viable    dags    retrieved    from    {}'.format(file_path))
        self.update_import_errors(session, dagbag)
        return []
    sync_time = datetime.now()
    for dag in dagbag.dags.values():
        models.DAG.sync_to_db(dag, dag.owner, sync_time)
    paused_dag_ids = [dag.dag_id for dag in dagbag.dags.values() if dag.is_paused]
    for dag_id in dagbag.dags:
        dag = dagbag.get_dag(dag_id)
        pickle_id = None
        if pickle_dags:
            pickle_id = dag.pickle(session).id
        task_ids = [task.task_id for task in dag.tasks]
        if (dag_id not in paused_dag_ids):
            simple_dags.append(SimpleDag(dag.dag_id, task_ids, dag.full_filepath, dag.concurrency, dag.is_paused, pickle_id))
    if (len(self.dag_ids) > 0):
        dags = [dag for dag in dagbag.dags.values() if ((dag.dag_id in self.dag_ids) and (dag.dag_id not in paused_dag_ids))]
    else:
        dags = [dag for dag in dagbag.dags.values() if ((not dag.parent_dag) and (dag.dag_id not in paused_dag_ids))]
    ti_keys_to_schedule = []
    self._process_dags(dagbag, dags, ti_keys_to_schedule)
    for ti_key in ti_keys_to_schedule:
        dag = dagbag.dags[ti_key[0]]
        task = dag.get_task(ti_key[1])
        ti = models.TaskInstance(task, ti_key[2])
        ti.refresh_from_db(session=session, lock_for_update=True)
        dep_context = DepContext(deps=QUEUE_DEPS, ignore_task_deps=True)
        if ti.are_dependencies_met(dep_context=dep_context, session=session, verbose=True):
            ti.state = State.SCHEDULED
        self.logger.info(u'Creating    /    updating    {}    in    ORM'.format(ti))
        session.merge(ti)
        session.commit()
    try:
        self.update_import_errors(session, dagbag)
    except Exception:
        self.logger.exception(u'Error    logging    import    errors!')
    try:
        dagbag.kill_zombies()
    except Exception:
        self.logger.exception(u'Error    killing    zombies!')
    return simple_dags
