[AIRFLOW-4844] Added optional is_paused_upon_creation argument for da…
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index b9c190ee78f37..2014587ade8ee 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -160,6 +160,10 @@ class DAG(BaseDag, LoggingMixin):
     :param access_control: Specify optional DAG-level permissions, e.g.,
         "{'role1': {'can_dag_read'}, 'role2': {'can_dag_read', 'can_dag_edit'}}"
     :type access_control: dict
+    :param is_paused_upon_creation: Specifies if the dag is paused when created for the first time.
+        If the dag exists already, this flag will be ignored. If this optional parameter
+        is not specified, the global config setting will be used.
+    :type is_paused_upon_creation: bool or None
     """
 
     def __init__(
@@ -186,7 +190,8 @@ def __init__(
         on_failure_callback: Optional[Callable] = None,
         doc_md: Optional[str] = None,
         params: Optional[Dict] = None,
-        access_control: Optional[Dict] = None
+        access_control: Optional[Dict] = None,
+        is_paused_upon_creation: Optional[bool] = None,
     ):
         self.user_defined_macros = user_defined_macros
         self.user_defined_filters = user_defined_filters
@@ -273,6 +278,7 @@ def __init__(
 
         self._old_context_manager_dags = []  # type: Iterable[DAG]
         self._access_control = access_control
+        self.is_paused_upon_creation = is_paused_upon_creation
 
         self._comps = {
             'dag_id',
@@ -1286,6 +1292,8 @@ def sync_to_db(self, owner=None, sync_time=None, session=None):
             DagModel).filter(DagModel.dag_id == self.dag_id).first()
         if not orm_dag:
             orm_dag = DagModel(dag_id=self.dag_id)
+            if self.is_paused_upon_creation is not None:
+                orm_dag.is_paused = self.is_paused_upon_creation
             self.log.info("Creating ORM DAG for %s", self.dag_id)
         orm_dag.fileloc = self.parent_dag.fileloc if self.is_subdag else self.fileloc
         orm_dag.is_subdag = self.is_subdag
diff --git a/tests/models/test_dag.py b/tests/models/test_dag.py
index 4830fcfdfb732..aaa0b40b2f748 100644
--- a/tests/models/test_dag.py
+++ b/tests/models/test_dag.py
@@ -972,3 +972,32 @@ def test_is_paused_subdag(self, mock_dag_bag):
         ).count()
 
         self.assertEqual(2, paused_dags)
+
+    def test_existing_dag_is_paused_upon_creation(self):
+        dag = DAG(
+            'dag'
+        )
+        session = settings.Session()
+        dag.sync_to_db(session=session)
+        orm_dag = session.query(DagModel).filter(DagModel.dag_id == 'dag').one()
+        self.assertFalse(orm_dag.is_paused)
+        dag = DAG(
+            'dag',
+            is_paused_upon_creation=True
+        )
+        dag.sync_to_db(session=session)
+        orm_dag = session.query(DagModel).filter(DagModel.dag_id == 'dag').one()
+        # Since the dag existed before, it should not follow the pause flag upon creation
+        self.assertFalse(orm_dag.is_paused)
+
+    def test_new_dag_is_paused_upon_creation(self):
+        dag = DAG(
+            'new_nonexisting_dag',
+            is_paused_upon_creation=True
+        )
+        session = settings.Session()
+        dag.sync_to_db(session=session)
+
+        orm_dag = session.query(DagModel).filter(DagModel.dag_id == 'new_nonexisting_dag').one()
+        # Since the dag didn't exist before, it should follow the pause flag upon creation
+        self.assertTrue(orm_dag.is_paused)

…gs to configure initial pause status upon creation.

Make sure you have checked _all_ steps below.

### Jira

- [ ] My PR addresses the following [Airflow-4844](https://issues.apache.org/jira/browse/AIRFLOW-4844)

### Description

- [ ] Here are some details about my PR:
1. Adding the feature to configure each dag object with the initial state of is_paused flag
2. The corresponding changes for python2 branch will be submitted once this review is finalized.


### Tests

- [ ] My PR adds the following unit tests:
1. Adding unit test to test existing dags in database not being affected with the flag.
2. Adding unit test to test new dags following the flag passed. (there are already unit tests testing backward compatibility of the the code change.)

### Commits

- [ ] My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
  1. Subject is separated from body by a blank line
  1. Subject is limited to 50 characters (not including Jira issue reference)
  1. Subject does not end with a period
  1. Subject uses the imperative mood ("add", not "adding")
  1. Body wraps at 72 characters
  1. Body explains "what" and "why", not "how"

### Documentation

- [ ] In case of new functionality, my PR adds documentation that describes how to use it.
  - All the public functions and the classes in the PR contain docstrings that explain what it does
  - If you implement backwards incompatible changes, please leave a note in the [Updating.md](https://github.com/apache/airflow/blob/master/UPDATING.md) so we can assign it to a appropriate release

### Code Quality

- [ ] Passes `flake8`
